You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2025/02/05 13:18:16 UTC
[PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
lhotari opened a new pull request, #23930:
URL: https://github.com/apache/pulsar/pull/23930
Fixes #23920
### Motivation
The current rate limiting implementation in Pulsar broker can cause connection timeouts when the state of the AsyncTokenBucket becomes invalid due to clock source issues in some virtualized environments where `System.nanoTime()` isn't strictly monotonic and consistent across multiple CPUs.
This PR addresses those problems.
### Modifications
1. **AsyncTokenBucket improvements**:
- Improved token calculation logic to prevent clock leaps backwards causing invalid state
- Improved token consumption logic while adding more test cases for the class (unrelated to the actual problem) to handle negative balance with eventual consistency:
- subtract pendingConsumedToken from new tokens before capping at bucket capacity
2. **DefaultMonotonicSnapshotClock enhancements**:
- Reimplemented using a single dedicated thread to reduce chances for clock leaps since a single CPU would be used in most cases unless the OS migrates the thread to run to another CPU
- Added logic to handle clock source leaps (backward/forward) gracefully
- Added required solution for use case where the consistent value of the clock is requested
- This is not on the hot path, so synchronization and Object monitor wait/notify/notifyAll provides a feasible solution.
3. **Rate limiter implementations**:
- Updated PublishRateLimiterImpl to use proper executor for unthrottling (EventLoopGroup will pick a new scheduler each time)
- Added error handling for producer unthrottling
- Modified DispatchRateLimiter and SubscribeRateLimiter to use the monotonic clock instance which handles the clock source issues
Added test coverage:
1. New AsyncTokenBucketTest cases:
- Fraction handling and leftover token retention
- Negative balance with eventual consistency
- Token bucket size limits
- Clock source lag scenarios
2. New DefaultMonotonicSnapshotClockTest:
- Time leap handling (backward/forward)
- Snapshot request validation
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1944343309
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
+ long clockValue = clockSource.getAsLong();
+
+ // Initialization
+ if (referenceClockSourceValue == 0) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = clockValue;
+ snapshotTickNanos = clockValue;
+ previousSnapshotTickNanos = clockValue;
+ return;
+ }
+
+ // calculate the duration since the reference clock source value
+ // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
+ // monotonic across all CPUs and leaps backward or forward
+ long durationSinceReference = clockValue - referenceClockSourceValue;
+ long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
+
+ // reset the reference clock source value if the clock source value leaps backward or forward
+ if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDelta
+ || newSnapshotTickNanos > previousSnapshotTickNanos + maxDelta) {
Review Comment:
Sure, I'll add some more comments to describe the logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#issuecomment-2642159133
> Perhaps this is not the root reason, many rate limiters use `System.nanoTime`, such as guava. We should avoid introducing a complex patch, even though the logic is clear.
>
> The minimal fix seems correct.
@nodece Just a reminder that this patch isn't complex based on code metrics. The problem domain is unusual and the feeling about complexity is mainly about unfamiliarity. This PR is still WIP and I'll try to get this into a reasonable state for final review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "rdhabalia (via GitHub)" <gi...@apache.org>.
rdhabalia commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1943667929
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
+ long clockValue = clockSource.getAsLong();
+
+ // Initialization
+ if (referenceClockSourceValue == 0) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = clockValue;
+ snapshotTickNanos = clockValue;
+ previousSnapshotTickNanos = clockValue;
+ return;
+ }
+
+ // calculate the duration since the reference clock source value
+ // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
+ // monotonic across all CPUs and leaps backward or forward
+ long durationSinceReference = clockValue - referenceClockSourceValue;
+ long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
+
+ // reset the reference clock source value if the clock source value leaps backward or forward
+ if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDelta
+ || newSnapshotTickNanos > previousSnapshotTickNanos + maxDelta) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = previousSnapshotTickNanos;
+ if (!snapshotRequested) {
+ // if the snapshot value is not requested, increment by the snapshot interval
+ baseSnapshotTickNanos += snapshotIntervalNanos;
+ }
+ newSnapshotTickNanos = baseSnapshotTickNanos;
+ }
+
+ // update snapshotTickNanos value if the new value is greater than the previous value
+ if (newSnapshotTickNanos > previousSnapshotTickNanos) {
+ snapshotTickNanos = newSnapshotTickNanos;
+ // store into a field so that we don't need to do a volatile read to find out the previous value
+ previousSnapshotTickNanos = newSnapshotTickNanos;
+ }
+ }
+
+ private void notifyAllTickUpdated() {
+ synchronized (tickUpdatedMonitor) {
+ // notify all threads that are waiting for the tick value to be updated
+ tickUpdatedMonitor.notifyAll();
+ }
+ }
+
+ public void requestUpdate() {
+ if (!running) {
+ // thread has stopped running, fallback to update the value directly without any optimizations
+ snapshotTickNanos = clockSource.getAsLong();
+ return;
+ }
+ synchronized (tickUpdatedMonitor) {
+ // notify the thread to stop waiting and update the tick value
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = true;
+ requestCount++;
+ tickUpdateDelayMonitor.notify();
+ }
+ // wait until the tick value has been updated
+ try {
+ tickUpdatedMonitor.wait();
+ } catch (InterruptedException e) {
+ currentThread().interrupt();
+ }
+ }
+ }
- private void snapshotLoop() {
- try {
- while (!Thread.currentThread().isInterrupted()) {
- updateSnapshotTickNanos();
+ @Override
+ public synchronized void start() {
+ super.start();
+ // wait until the thread is started and the tick value has been updated
+ synchronized (tickUpdatedMonitor) {
try {
- Thread.sleep(sleepMillis, sleepNanos);
+ tickUpdatedMonitor.wait();
} catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- break;
+ currentThread().interrupt();
}
}
- } catch (Throwable t) {
- // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
- // this is very unlikely to happen, but it's better to log it in any case
- LOG.error("Unexpected fatal error that stopped the clock.", t);
}
}
@Override
public void close() {
- thread.interrupt();
+ tickUpdaterThread.interrupt();
Review Comment:
can we terminate while loop with some variable `stop` and update it here. instead of depending on interrupting as it can lead leak if one changes the interrupt logic in future in previous method.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
+ long clockValue = clockSource.getAsLong();
+
+ // Initialization
+ if (referenceClockSourceValue == 0) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = clockValue;
+ snapshotTickNanos = clockValue;
+ previousSnapshotTickNanos = clockValue;
+ return;
+ }
+
+ // calculate the duration since the reference clock source value
+ // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
+ // monotonic across all CPUs and leaps backward or forward
+ long durationSinceReference = clockValue - referenceClockSourceValue;
+ long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
+
+ // reset the reference clock source value if the clock source value leaps backward or forward
+ if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDelta
+ || newSnapshotTickNanos > previousSnapshotTickNanos + maxDelta) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = previousSnapshotTickNanos;
+ if (!snapshotRequested) {
+ // if the snapshot value is not requested, increment by the snapshot interval
+ baseSnapshotTickNanos += snapshotIntervalNanos;
Review Comment:
here, we assume that `updateSnapshotTickNanos` gets invoked at every `snapshotIntervalNanos` time so, we are adding it to `baseSnapshotTickNanos`.
instead of if we pass `snapshotIntervalNanos` as a variable to this method then method could be less error prone and can be also used in unit-test if needed.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -134,7 +135,11 @@ private void unthrottleQueuedProducers(ScheduledExecutorService executor) {
// unthrottle as many producers as possible while there are token available
while ((throttlingDuration = calculateThrottlingDurationNanos()) == 0L
&& (producer = unthrottlingQueue.poll()) != null) {
- producer.decrementThrottleCount();
+ try {
+ producer.decrementThrottleCount();
+ } catch (Exception e) {
+ log.error("Failed to unthrottle producer {}", producer, e);
Review Comment:
worth to add topic name here
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
Review Comment:
it would be great if we can add unit-test for this method and try to get max code-coverage from unit-test as we can see high cyclomatic complexity in this method code.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
+ long clockValue = clockSource.getAsLong();
+
+ // Initialization
+ if (referenceClockSourceValue == 0) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = clockValue;
+ snapshotTickNanos = clockValue;
+ previousSnapshotTickNanos = clockValue;
+ return;
+ }
+
+ // calculate the duration since the reference clock source value
+ // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
+ // monotonic across all CPUs and leaps backward or forward
+ long durationSinceReference = clockValue - referenceClockSourceValue;
+ long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
+
+ // reset the reference clock source value if the clock source value leaps backward or forward
+ if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDelta
+ || newSnapshotTickNanos > previousSnapshotTickNanos + maxDelta) {
Review Comment:
btw, can we document what exactly we are doing with `baseSnapshotTickNanos` to explain the logic here as it's little hard to understand what is happening here?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
+ long clockValue = clockSource.getAsLong();
+
+ // Initialization
+ if (referenceClockSourceValue == 0) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = clockValue;
+ snapshotTickNanos = clockValue;
+ previousSnapshotTickNanos = clockValue;
+ return;
+ }
+
+ // calculate the duration since the reference clock source value
+ // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
+ // monotonic across all CPUs and leaps backward or forward
+ long durationSinceReference = clockValue - referenceClockSourceValue;
+ long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
+
+ // reset the reference clock source value if the clock source value leaps backward or forward
+ if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDelta
Review Comment:
now, as we are anyway, using single thread update, then can't we just avoid this complex check, and just directly use `System.nanoTime()` retrieving from the single CPU and use that incremented count?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1944296386
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
+ long clockValue = clockSource.getAsLong();
+
+ // Initialization
+ if (referenceClockSourceValue == 0) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = clockValue;
+ snapshotTickNanos = clockValue;
+ previousSnapshotTickNanos = clockValue;
+ return;
+ }
+
+ // calculate the duration since the reference clock source value
+ // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
+ // monotonic across all CPUs and leaps backward or forward
+ long durationSinceReference = clockValue - referenceClockSourceValue;
+ long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
+
+ // reset the reference clock source value if the clock source value leaps backward or forward
+ if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDelta
Review Comment:
A single thread could get migrated from one CPU to another. In certain virtualized environments, the counter can leap forward or backward. It can also happen on real multi-core/multi-socket hardware with TSC clock source when the hardware doesn't support CPU instructions which are required to address challenges in synchronizing the TSC values across CPUs. Some of the low level details is explained in ["TSC hardware"](https://docs.kernel.org/virt/kvm/x86/timekeeping.html#tsc-hardware).
The purpose of this class is to address any such problems where the time leap forward or backward and removing this logic would make this solution pointless.
Although this logic might seem complex, the problem is also a very low level problem. The logic is performant and has been validated with JMH benchmarks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1944342530
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/PublishRateLimiterImpl.java:
##########
@@ -134,7 +135,11 @@ private void unthrottleQueuedProducers(ScheduledExecutorService executor) {
// unthrottle as many producers as possible while there are token available
while ((throttlingDuration = calculateThrottlingDurationNanos()) == 0L
&& (producer = unthrottlingQueue.poll()) != null) {
- producer.decrementThrottleCount();
+ try {
+ producer.decrementThrottleCount();
+ } catch (Exception e) {
+ log.error("Failed to unthrottle producer {}", producer, e);
Review Comment:
The topic name is already included in producer's toString() method implementation.
https://github.com/apache/pulsar/blob/3d0625ba64294fb0fe7dafc27c7a34883b4be51b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Producer.java#L659-L662
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#issuecomment-2637700751
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/23930?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `90.09901%` with `10 lines` in your changes missing coverage. Please review.
> Project coverage is 74.26%. Comparing base [(`bbc6224`)](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`f719dbc`)](https://app.codecov.io/gh/apache/pulsar/commit/f719dbc70e875f42a1b93a5a9df9733e0cc7e0a9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 878 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/23930?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...lsar/broker/qos/DefaultMonotonicSnapshotClock.java](https://app.codecov.io/gh/apache/pulsar/pull/23930?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fqos%2FDefaultMonotonicSnapshotClock.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9xb3MvRGVmYXVsdE1vbm90b25pY1NuYXBzaG90Q2xvY2suamF2YQ==) | 89.74% | [7 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/23930?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../pulsar/broker/service/PublishRateLimiterImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/23930?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FPublishRateLimiterImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1B1Ymxpc2hSYXRlTGltaXRlckltcGwuamF2YQ==) | 66.66% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/23930?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/23930?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #23930 +/- ##
============================================
+ Coverage 73.57% 74.26% +0.68%
+ Complexity 32624 32261 -363
============================================
Files 1877 1853 -24
Lines 139502 143814 +4312
Branches 15299 16346 +1047
============================================
+ Hits 102638 106802 +4164
+ Misses 28908 28621 -287
- Partials 7956 8391 +435
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/23930/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/23930/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `26.68% <48.51%> (+2.10%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/23930/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `23.20% <48.51%> (-1.12%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/23930/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.79% <90.09%> (+0.94%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/23930?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...org/apache/pulsar/broker/qos/AsyncTokenBucket.java](https://app.codecov.io/gh/apache/pulsar/pull/23930?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fqos%2FAsyncTokenBucket.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9xb3MvQXN5bmNUb2tlbkJ1Y2tldC5qYXZh) | `93.05% <100.00%> (+0.09%)` | :arrow_up: |
| [...broker/service/persistent/DispatchRateLimiter.java](https://app.codecov.io/gh/apache/pulsar/pull/23930?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2Fpersistent%2FDispatchRateLimiter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3BlcnNpc3RlbnQvRGlzcGF0Y2hSYXRlTGltaXRlci5qYXZh) | `79.83% <100.00%> (+1.20%)` | :arrow_up: |
| [...roker/service/persistent/SubscribeRateLimiter.java](https://app.codecov.io/gh/apache/pulsar/pull/23930?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2Fpersistent%2FSubscribeRateLimiter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL3BlcnNpc3RlbnQvU3Vic2NyaWJlUmF0ZUxpbWl0ZXIuamF2YQ==) | `55.69% <100.00%> (+1.15%)` | :arrow_up: |
| [.../pulsar/broker/service/PublishRateLimiterImpl.java](https://app.codecov.io/gh/apache/pulsar/pull/23930?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FPublishRateLimiterImpl.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1B1Ymxpc2hSYXRlTGltaXRlckltcGwuamF2YQ==) | `91.30% <66.66%> (+5.15%)` | :arrow_up: |
| [...lsar/broker/qos/DefaultMonotonicSnapshotClock.java](https://app.codecov.io/gh/apache/pulsar/pull/23930?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fqos%2FDefaultMonotonicSnapshotClock.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9xb3MvRGVmYXVsdE1vbm90b25pY1NuYXBzaG90Q2xvY2suamF2YQ==) | `88.17% <89.74%> (+4.83%)` | :arrow_up: |
... and [1031 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/23930/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1944278326
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
+ long clockValue = clockSource.getAsLong();
+
+ // Initialization
+ if (referenceClockSourceValue == 0) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = clockValue;
+ snapshotTickNanos = clockValue;
+ previousSnapshotTickNanos = clockValue;
+ return;
+ }
+
+ // calculate the duration since the reference clock source value
+ // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
+ // monotonic across all CPUs and leaps backward or forward
+ long durationSinceReference = clockValue - referenceClockSourceValue;
+ long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
+
+ // reset the reference clock source value if the clock source value leaps backward or forward
+ if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDelta
+ || newSnapshotTickNanos > previousSnapshotTickNanos + maxDelta) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = previousSnapshotTickNanos;
+ if (!snapshotRequested) {
+ // if the snapshot value is not requested, increment by the snapshot interval
+ baseSnapshotTickNanos += snapshotIntervalNanos;
+ }
+ newSnapshotTickNanos = baseSnapshotTickNanos;
+ }
+
+ // update snapshotTickNanos value if the new value is greater than the previous value
+ if (newSnapshotTickNanos > previousSnapshotTickNanos) {
+ snapshotTickNanos = newSnapshotTickNanos;
+ // store into a field so that we don't need to do a volatile read to find out the previous value
+ previousSnapshotTickNanos = newSnapshotTickNanos;
+ }
+ }
+
+ private void notifyAllTickUpdated() {
+ synchronized (tickUpdatedMonitor) {
+ // notify all threads that are waiting for the tick value to be updated
+ tickUpdatedMonitor.notifyAll();
+ }
+ }
+
+ public void requestUpdate() {
+ if (!running) {
+ // thread has stopped running, fallback to update the value directly without any optimizations
+ snapshotTickNanos = clockSource.getAsLong();
+ return;
+ }
+ synchronized (tickUpdatedMonitor) {
+ // notify the thread to stop waiting and update the tick value
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = true;
+ requestCount++;
+ tickUpdateDelayMonitor.notify();
+ }
+ // wait until the tick value has been updated
+ try {
+ tickUpdatedMonitor.wait();
+ } catch (InterruptedException e) {
+ currentThread().interrupt();
+ }
+ }
+ }
- private void snapshotLoop() {
- try {
- while (!Thread.currentThread().isInterrupted()) {
- updateSnapshotTickNanos();
+ @Override
+ public synchronized void start() {
+ super.start();
+ // wait until the thread is started and the tick value has been updated
+ synchronized (tickUpdatedMonitor) {
try {
- Thread.sleep(sleepMillis, sleepNanos);
+ tickUpdatedMonitor.wait();
} catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- break;
+ currentThread().interrupt();
}
}
- } catch (Throwable t) {
- // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
- // this is very unlikely to happen, but it's better to log it in any case
- LOG.error("Unexpected fatal error that stopped the clock.", t);
}
}
@Override
public void close() {
- thread.interrupt();
+ tickUpdaterThread.interrupt();
Review Comment:
I don't think that there's a problem in using `.interrupt()` in this case. Internally, interrupt is a flag in the Thread class.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiter token bucket and clock consistency issues causing excessive throttling and connection timeouts [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1947557042
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucket.java:
##########
@@ -177,8 +186,13 @@ private long consumeTokensAndMaybeUpdateTokensBalance(long consumeTokens, boolea
pendingConsumedTokens.add(consumeTokens);
}
- // return Long.MIN_VALUE if the current value of tokens is unknown due to the eventual consistency
- return Long.MIN_VALUE;
+ if (forceUpdateTokens) {
+ // return the current tokens balance without updating the tokens and resetting the pendingConsumedTokens
+ return tokens - pendingConsumedTokens.sum();
Review Comment:
@nodece This code is now eliminated. I refactored AsyncTokenBucket to make the eventual consistency of it more understandable. There's javadoc in AsyncTokenBucketBuilder for the options to tune consistency since by default AsyncTokenBucket is by default eventually consistent so that the token balance is updated at most once during one "increment" which is configured with the resolutionNanos setting. By default it's 16 milliseconds.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiter token bucket and clock consistency issues causing excessive throttling and connection timeouts [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#issuecomment-2644261308
I fixed the flaky throttling tests and moved them to broker-api group. There were multiple issues in the flaky tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#issuecomment-2640930867
> Although clock sources aren't necessarily monotonic in all platforms, it seems that the root cause of this issue is could be different. The DefaultMonotonicSnapshotClock was originally added to address a performance bottleneck in calling `System.nanoTime()` on MacOS. This wasn't a significant bottleneck on Linux platform. The solution was to update the value in a single thread and add eventual consistency to the rate limiter. The problem seen in issue #23920 could occur when this thread gets starved and the updates are delayed. This PR already contains 2 mitigations to this problem by modifying the thread priority to `MAX_PRIORITY` and by modifying the logic how new tokens are added to the bucket when the token update happens. However, I'll need to find a solution where the problem would be properly avoided without adding contention since contention would cause the rate limiting solution to have a relatively high overhead.
I found a solution to the problem. It's possible to detect leaps backwards, but forward leaps should be unhandled since the tick updater thread could starve and time should be able to leap forward in that case when it finally gets updated.
Time leaping forward would cause issues with the changes in this PR. Before the changes made in this PR, there was an issue which caused negative tokens. I'll continue adding more tests to describe the scenario.
My assumption currently is that the minimal fix for #23920 is this diff:
```patch
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucket.java
index ac9a1f03e59..36edb1c5d8a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucket.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucket.java
@@ -159,16 +159,15 @@ public abstract class AsyncTokenBucket {
if (shouldUpdateTokensImmediately(currentNanos, forceUpdateTokens)) {
// calculate the number of new tokens since the last update
long newTokens = calculateNewTokensSinceLastUpdate(currentNanos);
- // calculate the total amount of tokens to consume in this update
// flush the pendingConsumedTokens by calling "sumThenReset"
- long totalConsumedTokens = consumeTokens + pendingConsumedTokens.sumThenReset();
+ long currentPendingConsumedTokens = pendingConsumedTokens.sumThenReset();
// update the tokens and return the current token value
return TOKENS_UPDATER.updateAndGet(this,
- currentTokens ->
- // after adding new tokens, limit the tokens to the capacity
- Math.min(currentTokens + newTokens, getCapacity())
- // subtract the consumed tokens
- - totalConsumedTokens);
+ // after adding new tokens subtract the pending consumed tokens and
+ // limit the tokens to the capacity of the bucket
+ currentTokens -> Math.min(currentTokens + newTokens - currentPendingConsumedTokens, getCapacity())
+ // subtract the consumed tokens
+ - consumeTokens);
} else {
// eventual consistent fast path, tokens are not updated immediately
```
The problem with the previous solution was that it would first add new tokens and cap it to the capacity. With the eventually consistency logic in AsyncTokenBucket, that's a wrong solution. I'll provide a separate explanation for this in later updates.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiter token bucket and clock consistency issues causing connection timeouts [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1946372607
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
+ long clockValue = clockSource.getAsLong();
+
+ // Initialization
+ if (referenceClockSourceValue == 0) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = clockValue;
+ snapshotTickNanos = clockValue;
+ previousSnapshotTickNanos = clockValue;
+ return;
+ }
+
+ // calculate the duration since the reference clock source value
+ // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
+ // monotonic across all CPUs and leaps backward or forward
+ long durationSinceReference = clockValue - referenceClockSourceValue;
+ long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
+
+ // reset the reference clock source value if the clock source value leaps backward or forward
+ if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDelta
+ || newSnapshotTickNanos > previousSnapshotTickNanos + maxDelta) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = previousSnapshotTickNanos;
+ if (!snapshotRequested) {
+ // if the snapshot value is not requested, increment by the snapshot interval
+ baseSnapshotTickNanos += snapshotIntervalNanos;
Review Comment:
The implementation has been refactored. I kept the other classes as internal classes since they are not intented to be used separately from DefaultMonotonicSnapshotClock. It makes it slightly easier to read the code when you don't have to navigate to a separate class file. I have added a unit test also for the leap detection behavior.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1944340948
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
Review Comment:
I added a few tests and the line coverage is now 97% (73/75) and branch coverage is 90% (18/20) with the current DefaultMonotonicSnapshotClock. There's only 2 lines and 2 branches that aren't covered which are lines 196 and 209 where there's handling for InterruptedException. Adding test coverage for those lines isn't reasonable, so this means that max code coverage is reached.
Based on metrics, the complexity of the of method code isn't high since methods are fairly small.
<img width="1259" alt="image" src="https://github.com/user-attachments/assets/3bf121c1-694b-4a61-bb98-90abc6710dfd" />
for example, PersistentTopic contains methods which have high complexity according to metrics:
<img width="1229" alt="image" src="https://github.com/user-attachments/assets/970976ce-4a4c-408f-9551-5de640dabfcc" />
Metrics calculated with IntelliJ plugin "[MetricsReloaded](https://plugins.jetbrains.com/plugin/93-metricsreloaded)"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#issuecomment-2640613637
Although clock sources aren't necessarily monotonic in all platforms, it seems that the root cause of this issue is could be different.
The DefaultMonotonicSnapshotClock was originally added to address a performance bottleneck in calling `System.nanoTime()` on MacOS. This wasn't a significant bottleneck on Linux platform. The solution was to update the value in a single thread and add eventual consistency to the rate limiter. The problem seen in issue #23920 could occur when this thread gets starved and the updates are delayed. This PR already contains 2 mitigations to this problem by modifying the thread priority to `MAX_PRIORITY` and by modifying the logic how new tokens are added to the bucket when the token update happens.
However, I'll need to find a solution where the problem would be properly avoided without adding contention since contention would cause the rate limiting solution to have a relatively high overhead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiter token bucket and clock consistency issues causing excessive throttling and connection timeouts [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1947494686
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/AsyncTokenBucket.java:
##########
@@ -177,8 +186,13 @@ private long consumeTokensAndMaybeUpdateTokensBalance(long consumeTokens, boolea
pendingConsumedTokens.add(consumeTokens);
}
- // return Long.MIN_VALUE if the current value of tokens is unknown due to the eventual consistency
- return Long.MIN_VALUE;
+ if (forceUpdateTokens) {
+ // return the current tokens balance without updating the tokens and resetting the pendingConsumedTokens
+ return tokens - pendingConsumedTokens.sum();
Review Comment:
```suggestion
return tokens - pendingConsumedTokens.sumThenReset();
```
> resetting the pendingConsumedTokens
Should be `sumThenReset()`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiter token bucket and clock consistency issues causing excessive throttling and connection timeouts [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #23930:
URL: https://github.com/apache/pulsar/pull/23930
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiter token bucket and clock consistency issues causing connection timeouts [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1946373881
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
+ long clockValue = clockSource.getAsLong();
+
+ // Initialization
+ if (referenceClockSourceValue == 0) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = clockValue;
+ snapshotTickNanos = clockValue;
+ previousSnapshotTickNanos = clockValue;
+ return;
+ }
+
+ // calculate the duration since the reference clock source value
+ // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
+ // monotonic across all CPUs and leaps backward or forward
+ long durationSinceReference = clockValue - referenceClockSourceValue;
+ long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
+
+ // reset the reference clock source value if the clock source value leaps backward or forward
+ if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDelta
Review Comment:
I have updated the PR description with the rationale. As mentioned before, this code isn't complex based on code metrics. The problem domain is unusual and the feeling about complexity is mainly about unfamiliarity.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#issuecomment-2641930979
> `System.nanoTime()` isn't strictly monotonic and consistent across multiple CPUs.
Perhaps this is not the root reason, many rate limiters use `System.nanoTime`, such as guava. We should avoid introducing a complex patch, even though the logic is clear.
The minimal fix seems correct.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiter token bucket and clock consistency issues causing connection timeouts [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1946376132
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
Review Comment:
Code coverage is maxed currently for DefaultMonotonicSnapshotClock. 100% branch coverage and 97% line coverage (everything else except `currentThread().interrupt()` lines in catching `InterruptedException`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1944356918
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
+ long clockValue = clockSource.getAsLong();
+
+ // Initialization
+ if (referenceClockSourceValue == 0) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = clockValue;
+ snapshotTickNanos = clockValue;
+ previousSnapshotTickNanos = clockValue;
+ return;
+ }
+
+ // calculate the duration since the reference clock source value
+ // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
+ // monotonic across all CPUs and leaps backward or forward
+ long durationSinceReference = clockValue - referenceClockSourceValue;
+ long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
+
+ // reset the reference clock source value if the clock source value leaps backward or forward
+ if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDelta
+ || newSnapshotTickNanos > previousSnapshotTickNanos + maxDelta) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = previousSnapshotTickNanos;
+ if (!snapshotRequested) {
+ // if the snapshot value is not requested, increment by the snapshot interval
+ baseSnapshotTickNanos += snapshotIntervalNanos;
Review Comment:
I'll refactor out the separate concern to detect leaps in the clock source values and normalize the output values to a separate class since that could help unit test it separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiter token bucket and clock consistency issues causing excessive throttling and connection timeouts [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#issuecomment-2643604500
Checking test failures in SubscriptionMessageDispatchThrottlingTest
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiter token bucket and clock consistency issues causing connection timeouts [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#discussion_r1946374506
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/qos/DefaultMonotonicSnapshotClock.java:
##########
@@ -45,45 +49,170 @@ public DefaultMonotonicSnapshotClock(long snapshotIntervalNanos, LongSupplier cl
this.sleepMillis = TimeUnit.NANOSECONDS.toMillis(snapshotIntervalNanos);
this.sleepNanos = (int) (snapshotIntervalNanos - TimeUnit.MILLISECONDS.toNanos(sleepMillis));
this.clockSource = clockSource;
- updateSnapshotTickNanos();
- thread = new Thread(this::snapshotLoop, getClass().getSimpleName() + "-update-loop");
- thread.setDaemon(true);
- thread.start();
+ this.snapshotIntervalNanos = snapshotIntervalNanos;
+ tickUpdaterThread = new TickUpdaterThread();
+ tickUpdaterThread.start();
}
/** {@inheritDoc} */
@Override
public long getTickNanos(boolean requestSnapshot) {
if (requestSnapshot) {
- updateSnapshotTickNanos();
+ tickUpdaterThread.requestUpdate();
}
return snapshotTickNanos;
}
- private void updateSnapshotTickNanos() {
- snapshotTickNanos = clockSource.getAsLong();
- }
+ /**
+ * A thread that updates snapshotTickNanos value periodically with a configured interval.
+ * The thread is started when the DefaultMonotonicSnapshotClock is created and runs until the close method is
+ * called.
+ * A single thread is used to read the clock source value since on some hardware of virtualized platforms,
+ * System.nanoTime() isn't strictly monotonic across all CPUs. Reading by a single thread will improve the
+ * stability of the read value since a single thread is scheduled on a single CPU. If the thread is migrated
+ * to another CPU, the clock source value might leap backward or forward, but logic in this class will handle it.
+ */
+ private class TickUpdaterThread extends Thread {
+ private final Object tickUpdateDelayMonitor = new Object();
+ private final Object tickUpdatedMonitor = new Object();
+ private final long maxDelta;
+ private long referenceClockSourceValue;
+ private long baseSnapshotTickNanos;
+ private long previousSnapshotTickNanos;
+ private volatile boolean running;
+ private boolean tickUpdateDelayMonitorNotified;
+ private long requestCount;
+
+ TickUpdaterThread() {
+ super(DefaultMonotonicSnapshotClock.class.getSimpleName() + "-update-loop");
+ // set as daemon thread so that it doesn't prevent the JVM from exiting
+ setDaemon(true);
+ // set the highest priority
+ setPriority(MAX_PRIORITY);
+ this.maxDelta = 2 * snapshotIntervalNanos;
+ }
+
+ @Override
+ public void run() {
+ try {
+ running = true;
+ long updatedForRequestCount = -1;
+ while (!isInterrupted()) {
+ try {
+ boolean snapshotRequested = false;
+ // sleep for the configured interval on a monitor that can be notified to stop the sleep
+ // and update the tick value immediately. This is used in requestUpdate method.
+ synchronized (tickUpdateDelayMonitor) {
+ tickUpdateDelayMonitorNotified = false;
+ // only wait if no explicit request has been made since the last update
+ if (requestCount == updatedForRequestCount) {
+ // if no request has been made, sleep for the configured interval
+ tickUpdateDelayMonitor.wait(sleepMillis, sleepNanos);
+ snapshotRequested = tickUpdateDelayMonitorNotified;
+ }
+ updatedForRequestCount = requestCount;
+ }
+ updateSnapshotTickNanos(snapshotRequested);
+ notifyAllTickUpdated();
+ } catch (InterruptedException e) {
+ interrupt();
+ break;
+ }
+ }
+ } catch (Throwable t) {
+ // report unexpected error since this would be a fatal error when the clock doesn't progress anymore
+ // this is very unlikely to happen, but it's better to log it in any case
+ LOG.error("Unexpected fatal error that stopped the clock.", t);
+ } finally {
+ LOG.info("DefaultMonotonicSnapshotClock's TickUpdaterThread stopped. {},tid={}", this, getId());
+ running = false;
+ notifyAllTickUpdated();
+ }
+ }
+
+ private void updateSnapshotTickNanos(boolean snapshotRequested) {
+ long clockValue = clockSource.getAsLong();
+
+ // Initialization
+ if (referenceClockSourceValue == 0) {
+ referenceClockSourceValue = clockValue;
+ baseSnapshotTickNanos = clockValue;
+ snapshotTickNanos = clockValue;
+ previousSnapshotTickNanos = clockValue;
+ return;
+ }
+
+ // calculate the duration since the reference clock source value
+ // so that the snapshot value is always increasing and tolerates it when the clock source is not strictly
+ // monotonic across all CPUs and leaps backward or forward
+ long durationSinceReference = clockValue - referenceClockSourceValue;
+ long newSnapshotTickNanos = baseSnapshotTickNanos + durationSinceReference;
+
+ // reset the reference clock source value if the clock source value leaps backward or forward
+ if (newSnapshotTickNanos < previousSnapshotTickNanos - maxDelta
+ || newSnapshotTickNanos > previousSnapshotTickNanos + maxDelta) {
Review Comment:
The code has been refactored and there are comments to explain the logic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiting causing connections to time out [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#issuecomment-2642482616
I have also added a JMH benchmark for DefaultMonotonicSnapshotClock. Results on Apple Mac M3 Max:
```
Benchmark Mode Cnt Score Error Units
DefaultMonotonicSnapshotClockBenchmark.getTickNanos001Threads thrpt 3 2033254223.572 ± 41581820.145 ops/s
DefaultMonotonicSnapshotClockBenchmark.getTickNanos010Threads thrpt 3 18719597588.557 ± 2124798619.451 ops/s
DefaultMonotonicSnapshotClockBenchmark.getTickNanos100Threads thrpt 3 21049656355.267 ± 22701352214.776 ops/s
DefaultMonotonicSnapshotClockBenchmark.getTickNanosRequestSnapshot001Threads thrpt 3 70079972.669 ± 1667215.446 ops/s
DefaultMonotonicSnapshotClockBenchmark.getTickNanosRequestSnapshot010Threads thrpt 3 22865514.539 ± 1216715.428 ops/s
DefaultMonotonicSnapshotClockBenchmark.getTickNanosRequestSnapshot100Threads thrpt 3 16422340.515 ± 2346690.916 ops/s
```
With the consistent mode (`requestSnapshot=true`), the throughput is about 16.4 M ops/s with 100 threads so the implementation scales very well for it's purpose. The AsyncTokenBucket implementation has a solution to minimize the calls for a consistent view to up to 1 call for a snapshot interval so this will scale very well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][broker] Fix rate limiter token bucket and clock consistency issues causing connection timeouts [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23930:
URL: https://github.com/apache/pulsar/pull/23930#issuecomment-2642637697
This PR is now ready for final review. Please PTAL @rdhabalia @nodece. I'd appreciate accurate feedback about possible concerns about complexity since the based on code metrics this PR doesn't contain complex code. The problem domain is unusual and the feeling about complexity is mainly about unfamiliarity. I'll be happy to improve the description if this is the case. I'm confident that this PR is essential and important for the [PIP-322 Rate Limiting implementation](https://github.com/apache/pulsar/blob/master/pip/pip-322.md) that has been included in Pulsar since Pulsar 3.2.0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org