You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@lucene.apache.org by "nipunbatra8 (via GitHub)" <gi...@apache.org> on 2025/07/17 22:14:17 UTC
[PR] [Draft] Add BandwidthCappedMergeScheduler: A Global Bandwidth-Limiting MergeScheduler [lucene]
nipunbatra8 opened a new pull request, #14964:
URL: https://github.com/apache/lucene/pull/14964
# Add BandwidthCappedMergeScheduler: A Global Bandwidth-Limiting MergeScheduler
This draft PR introduces a prototype `BandwidthCappedMergeScheduler`, which extends `ConcurrentMergeScheduler` to enforce a global bandwidth cap on merge operations across all active merges within an `IndexWriter`. The scheduler is inspired by the issue/discussion in [lucene#14148](https://github.com/apache/lucene/issues/14148) and feedback from @mikemccand and others. The motivation is to provide a simple, global bandwidth cap for merge operations, especially useful during "update storms" or "war time" scenarios where aggressive merging can cause page faults.
## Implementation
- **Introduces `BandwidthCappedMergeScheduler` in `lucene.index`**
- **Extends `ConcurrentMergeScheduler` to reuse merge management and threading**
- **Implements a global bandwidth cap by dividing a configurable MB/s limit among all active merges**
- **Overrides `updateMergeThreads()` to dynamically adjust per-merge IO rates**
Initially, I experimented with skipping merges in the scheduler (by aborting or refusing to run merges that would exceed the bandwidth cap). However, this approach proved problematic:
- **Resource Leaks:** By the time the scheduler can skip a merge, resources/files may already be allocated, leading to leaks or inconsistent state.
- **Merge Policy Coordination:** Skipping merges late in the scheduler can result in better merges being blocked or waiting indefinitely, and it’s too late to make an informed decision without communicating with the `IndexWriter` or the `MergePolicy`.
- **Complexity:** Trying to abort merges at the scheduler level introduces complexity and potential for subtle bugs, especially around resource cleanup and merge queue management.
## CMS IO Rate Limiter?
- **ConcurrentMergeScheduler (CMS):** Applies adaptive IO throttling to each merge thread individually, so total bandwidth may exceed a set limit when multiple merges run.
- **BandwidthCappedMergeScheduler:** Divides a single global bandwidth cap among all active merges, ensuring their combined IO rate never exceeds the configured maximum.
### Should We Integrate with CMS IO Rate Limiter?
One open question: should we integrate CMS’s adaptive IO rate logic into this scheduler as well, or something like switch to the global cap only during "war time", such as:
- Use CMS’s adaptive per-merge throttling during "peace time" (normal operation).
- Switch to the global bandwidth cap when the system detects "war time" (e.g., max merges/threads reached, or backlog detected).
## Future Improvements
For now, the implementation is intentionally simple. However, there are several ways to make it more efficient and fair:
1. **Sort merges by size:** Prioritize smaller merges or allocate bandwidth proportionally to merge size.
2. **Priority-based throttling:** Give more bandwidth to urgent merges (e.g., forceMerge, high delete reclaim).
3. **Dynamic feedback:** Adjust cap/rates based on system load actually consumed by merges
4. **Merge policy integration:** Communicate bandwidth usage back to the `MergePolicy` to influence which merges are selected, or even skip merges earlier in the process.
## Testing
- I have tested this scheduler by replacing CMS in `LuceneTestCase.java` and running the full test suite.
- Next, I plan to edit `NRTPerfTest` to simulate update storms and then generate segment traces to visualize bandwidth usage and merge behavior. Hopefully see a less spiky segment tracing graph.
- **Request for feedback:** Are there other suggestions for tests or benchmarks to monitor this?
## Next Steps
- **Discuss with the community**
- **Testing to visualize results as mentioned in the Testing section**
- **Efficiency improvements:** Explore more advanced bandwidth allocation strategies as described above in future improvements.
- **Merge policy integration:** Investigate ways to skip or prioritize merges earlier, possibly by editing `IndexWriter` or implementing a `BandwidthAwareTieredMergePolicy` that can respond to bandwidth constraints in real time. Thoughts on this?
---
**Thanks for reviewing! Looking forward to feedback, suggestions, and further discussion. This PR is opened to propose and evaluate a bandwidth-capped merge scheduler design, and to gather feedback for further development.**
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "shubhamsrkdev (via GitHub)" <gi...@apache.org>.
shubhamsrkdev commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2306919441
##########
lucene/sandbox/src/java/org/apache/lucene/sandbox/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+import org.apache.lucene.index.MergeScheduler;
+
+/**
+ * A {@link MergeScheduler} that caps total IO write bandwidth across all running merges to a
+ * specified max MB/sec bandwidth.
+ *
+ * @lucene.experimental
+ */
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Global bandwidth cap in MB/s. Mutable so that updates are applied live. */
+ private double bandwidthMbPerSec;
+
+ /** Create a scheduler with a required global bandwidth cap in MB/s. */
+ public BandwidthCappedMergeScheduler(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
Review Comment:
Imo, this could just be :
`bandwidthMbPerSec <= 0.0`
This would be more readable.
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "shubhamsrkdev (via GitHub)" <gi...@apache.org>.
shubhamsrkdev commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2306995447
##########
lucene/sandbox/src/java/org/apache/lucene/sandbox/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+import org.apache.lucene.index.MergeScheduler;
+
+/**
+ * A {@link MergeScheduler} that caps total IO write bandwidth across all running merges to a
+ * specified max MB/sec bandwidth.
+ *
+ * @lucene.experimental
+ */
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Global bandwidth cap in MB/s. Mutable so that updates are applied live. */
+ private double bandwidthMbPerSec;
+
+ /** Create a scheduler with a required global bandwidth cap in MB/s. */
+ public BandwidthCappedMergeScheduler(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
+ || Double.isNaN(bandwidthMbPerSec)
+ || Double.isInfinite(bandwidthMbPerSec)) {
+ throw new IllegalArgumentException("bandwidthMbPerSec must be a finite positive value");
+ }
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ }
+
+ /**
+ * Auto IO throttling is managed by this scheduler's bandwidth cap. Enabling parent CMS IO
+ * throttling is ignored.
+ */
+ @Override
+ public synchronized void enableAutoIOThrottle() {
+ if (verbose()) {
+ message("Ignoring enableAutoIOThrottle; using bandwidth cap instead");
+ }
+ // Intentionally no-op
+ }
+
+ /** Ensure auto IO throttling remains disabled. */
+ @Override
+ public synchronized void disableAutoIOThrottle() {
+ // Make sure parent state is disabled if it was somehow enabled earlier
+ super.disableAutoIOThrottle();
+ }
+
+ /** Always returns false since CMS auto IO throttling is disabled for this scheduler. */
+ @Override
+ public synchronized boolean getAutoIOThrottle() {
+ return false;
+ }
+
+ /** Get the global bandwidth cap in MB/s */
+ public double getMaxMbPerSec() {
+ return bandwidthMbPerSec;
+ }
+
+ /**
+ * Set the global bandwidth cap in MB/s.
+ *
+ * <p>This setting is live: merges that are already running will be adjusted to the new per-merge
+ * rate derived from this cap, without requiring a restart.
+ *
+ * @param bandwidthMbPerSec the new global bandwidth cap in MB/s; must be finite and > 0
+ */
+ public synchronized void setMaxMbPerSec(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
+ || Double.isNaN(bandwidthMbPerSec)
+ || Double.isInfinite(bandwidthMbPerSec)) {
+ throw new IllegalArgumentException("bandwidthMbPerSec must be a finite positive value");
+ }
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+
+ // Use the effective max thread count to avoid counting threads that CMS has paused
+ int effectiveMaxThreads = getMaxThreadCount();
+ if (effectiveMaxThreads == ConcurrentMergeScheduler.AUTO_DETECT_MERGES_AND_THREADS) {
+ int coreCount = Runtime.getRuntime().availableProcessors();
+ effectiveMaxThreads = Math.max(1, coreCount / 2);
+ }
+ int divisor = Math.min(effectiveMaxThreads, activeMerges);
+
+ double perMergeRate;
+ if (divisor > 0) {
+ perMergeRate = Math.max(0.1, bandwidthMbPerSec / divisor); // Use 0.1 MB/s minimum
+ } else {
+ perMergeRate = Double.POSITIVE_INFINITY;
+ }
+
+ // Apply the calculated rate limit to each active merge thread without unpausing paused threads
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ double currentRate = mergeThread.getRateLimiter().getMBPerSec();
+ if (currentRate > 0.0) { // Only update if not paused by parent CMS (above soft limit)
+ mergeThread.getRateLimiter().setMBPerSec(perMergeRate);
+ }
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthMbPerSec="
+ + bandwidthMbPerSec
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ }
+
+ @Override
+ public void run() {
+ long startTimeNS = System.nanoTime();
+ OneMerge merge = getMerge();
+ try {
+ if (verbose()) {
+ message(
Review Comment:
nit: Shall we use string.format here?
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "shubhamsrkdev (via GitHub)" <gi...@apache.org>.
shubhamsrkdev commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2306993663
##########
lucene/sandbox/src/java/org/apache/lucene/sandbox/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+import org.apache.lucene.index.MergeScheduler;
+
+/**
+ * A {@link MergeScheduler} that caps total IO write bandwidth across all running merges to a
+ * specified max MB/sec bandwidth.
+ *
+ * @lucene.experimental
+ */
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Global bandwidth cap in MB/s. Mutable so that updates are applied live. */
+ private double bandwidthMbPerSec;
+
+ /** Create a scheduler with a required global bandwidth cap in MB/s. */
+ public BandwidthCappedMergeScheduler(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
+ || Double.isNaN(bandwidthMbPerSec)
+ || Double.isInfinite(bandwidthMbPerSec)) {
+ throw new IllegalArgumentException("bandwidthMbPerSec must be a finite positive value");
+ }
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ }
+
+ /**
+ * Auto IO throttling is managed by this scheduler's bandwidth cap. Enabling parent CMS IO
+ * throttling is ignored.
+ */
+ @Override
+ public synchronized void enableAutoIOThrottle() {
+ if (verbose()) {
+ message("Ignoring enableAutoIOThrottle; using bandwidth cap instead");
+ }
+ // Intentionally no-op
+ }
+
+ /** Ensure auto IO throttling remains disabled. */
+ @Override
+ public synchronized void disableAutoIOThrottle() {
+ // Make sure parent state is disabled if it was somehow enabled earlier
+ super.disableAutoIOThrottle();
+ }
+
+ /** Always returns false since CMS auto IO throttling is disabled for this scheduler. */
+ @Override
+ public synchronized boolean getAutoIOThrottle() {
+ return false;
+ }
+
+ /** Get the global bandwidth cap in MB/s */
+ public double getMaxMbPerSec() {
+ return bandwidthMbPerSec;
+ }
+
+ /**
+ * Set the global bandwidth cap in MB/s.
+ *
+ * <p>This setting is live: merges that are already running will be adjusted to the new per-merge
+ * rate derived from this cap, without requiring a restart.
+ *
+ * @param bandwidthMbPerSec the new global bandwidth cap in MB/s; must be finite and > 0
+ */
+ public synchronized void setMaxMbPerSec(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
+ || Double.isNaN(bandwidthMbPerSec)
+ || Double.isInfinite(bandwidthMbPerSec)) {
+ throw new IllegalArgumentException("bandwidthMbPerSec must be a finite positive value");
+ }
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+
+ // Use the effective max thread count to avoid counting threads that CMS has paused
+ int effectiveMaxThreads = getMaxThreadCount();
+ if (effectiveMaxThreads == ConcurrentMergeScheduler.AUTO_DETECT_MERGES_AND_THREADS) {
+ int coreCount = Runtime.getRuntime().availableProcessors();
+ effectiveMaxThreads = Math.max(1, coreCount / 2);
+ }
+ int divisor = Math.min(effectiveMaxThreads, activeMerges);
+
+ double perMergeRate;
+ if (divisor > 0) {
+ perMergeRate = Math.max(0.1, bandwidthMbPerSec / divisor); // Use 0.1 MB/s minimum
+ } else {
+ perMergeRate = Double.POSITIVE_INFINITY;
+ }
+
+ // Apply the calculated rate limit to each active merge thread without unpausing paused threads
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ double currentRate = mergeThread.getRateLimiter().getMBPerSec();
+ if (currentRate > 0.0) { // Only update if not paused by parent CMS (above soft limit)
+ mergeThread.getRateLimiter().setMBPerSec(perMergeRate);
+ }
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthMbPerSec="
+ + bandwidthMbPerSec
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ }
+
+ @Override
+ public void run() {
+ long startTimeNS = System.nanoTime();
+ OneMerge merge = getMerge();
+ try {
+ if (verbose()) {
+ message(
+ "Starting bandwidth-capped merge: (estimatedMergeMB="
+ + merge.estimatedMergeBytes / (1024.0 * 1024.0)
+ + " MB)");
+ }
+ super.run(); // IO throttling is handled by the RateLimiter
+ } finally {
+ long durationNS = System.nanoTime() - startTimeNS;
+ if (verbose()) {
+ double durationMS = durationNS / 1_000_000.0;
+ double mbPerSec = merge.estimatedMergeBytes / (1024.0 * 1024.0) / (durationMS / 1000.0);
+ message(
+ "Merge completed: "
+ + merge.estimatedMergeBytes / (1024.0 * 1024.0)
+ + " MB in "
+ + String.format(java.util.Locale.US, "%.1f", durationMS)
Review Comment:
Is there a reason this is `Locale.US` and not `Locale.ROOT`? Same for below.
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "nipunbatra8 (via GitHub)" <gi...@apache.org>.
nipunbatra8 commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2252161206
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
+ : Double.POSITIVE_INFINITY;
+
+ // Apply the calculated rate limit to each active merge thread
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ mergeThread.rateLimiter.setMBPerSec(perMergeRate);
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthRateBucket="
+ + bandwidthRateBucket
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+ private final double mergeBandwidthMB;
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ this.mergeBandwidthMB = merge.estimatedMergeBytes / (1024.0 * 1024.0);
+ }
+
+ @Override
+ public void run() {
+ long startTime = System.currentTimeMillis();
+ try {
+ if (verbose()) {
+ message(
+ "Starting bandwidth-capped merge: "
+ + getSegmentName(merge)
+ + " (estimated="
+ + mergeBandwidthMB
+ + " MB)");
+ }
+ super.run(); // IO throttling is handled by the RateLimiter
+ } finally {
+ long duration = System.currentTimeMillis() - startTime;
+ if (verbose()) {
+ double mbPerSec = mergeBandwidthMB / Math.max(duration / 1000.0, 0.001);
Review Comment:
I have the 0.001 to avoid division by zero problems in case the duration was computed as 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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "nipunbatra8 (via GitHub)" <gi...@apache.org>.
nipunbatra8 commented on PR #14964:
URL: https://github.com/apache/lucene/pull/14964#issuecomment-3152551802
@stefanvodita The original issue is that update storms replicate many large segments out to searchers rapidly due to a lot of merging causing paging storms when they finally load a massive snapshot. So rather than preventing large segments by skipping merges, I control the rate at which merges are being run by throttling merge I/O bandwidth, spreading out segment creation over time to prevent the delta of the snapshot "storm" of large segments that overwhelms searchers.
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "shubhamsrkdev (via GitHub)" <gi...@apache.org>.
shubhamsrkdev commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2307039991
##########
lucene/sandbox/src/test/org/apache/lucene/sandbox/index/TestBandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import com.carrotsearch.randomizedtesting.generators.RandomStrings;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.document.TextField;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.index.LogDocMergePolicy;
+import org.apache.lucene.index.MergePolicy;
+import org.apache.lucene.index.MergeScheduler;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.tests.util.LuceneTestCase;
+import org.apache.lucene.util.InfoStream;
+
+/** Tests for {@link BandwidthCappedMergeScheduler}. */
+public class TestBandwidthCappedMergeScheduler extends LuceneTestCase {
+
+ public void testBasicFunctionality() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler(100.0);
+
+ // Test initial value
+ assertEquals(100.0, scheduler.getMaxMbPerSec(), 0.001);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testBandwidthConfiguration() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler(50.0);
+
+ // Test setting valid bandwidth rates
+ scheduler.setMaxMbPerSec(200.0);
+ assertEquals(200.0, scheduler.getMaxMbPerSec(), 0.001);
+
+ scheduler.setMaxMbPerSec(10.0);
+ assertEquals(10.0, scheduler.getMaxMbPerSec(), 0.001);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testInvalidBandwidthConfiguration() throws Exception {
+ // Test invalid constructor parameter
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ new BandwidthCappedMergeScheduler(0.0);
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ new BandwidthCappedMergeScheduler(-1.0);
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ new BandwidthCappedMergeScheduler(Double.NaN);
+ });
+
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler(100.0);
+
+ // Test invalid setMaxMbPerSec values
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setMaxMbPerSec(0.0);
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setMaxMbPerSec(-1.0);
+ });
+
+ scheduler.close();
+ }
+
+ public void testToString() throws Exception {
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler(50.0);
+ String str = scheduler.toString();
+
+ assertTrue(str.contains("BandwidthCappedMergeScheduler"));
+ assertTrue(str.contains("bandwidthMbPerSec"));
+ assertTrue(str.contains("MB/s"));
+ assertTrue(str.contains("50.0"));
+
+ scheduler.close();
+ }
+
+ public void testWithIndexWriter() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler(100.0);
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2); // Force frequent flushes
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ // Add some documents to potentially trigger merges
+ for (int i = 0; i < 20; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(
+ new TextField(
+ "content",
+ "content " + i + " " + RandomStrings.randomRealisticUnicodeOfLength(random(), 100),
+ Field.Store.YES));
+ writer.addDocument(doc);
+
+ if (i % 5 == 0) {
+ writer.commit(); // Force segments
+ }
+ }
+
+ writer.forceMerge(1); // This should trigger merges
+ }
+
+ // The scheduler should have been used
+ assertTrue("Scheduler should have been initialized", scheduler.getMaxMbPerSec() > 0);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testBandwidthDistributionAmongMerges() throws IOException {
Review Comment:
This test has a missing assertion, what are we asserting here?
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "shubhamsrkdev (via GitHub)" <gi...@apache.org>.
shubhamsrkdev commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2307007915
##########
lucene/sandbox/src/test/org/apache/lucene/sandbox/index/TestBandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import com.carrotsearch.randomizedtesting.generators.RandomStrings;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.document.TextField;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.index.LogDocMergePolicy;
+import org.apache.lucene.index.MergePolicy;
+import org.apache.lucene.index.MergeScheduler;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.tests.util.LuceneTestCase;
+import org.apache.lucene.util.InfoStream;
+
+/** Tests for {@link BandwidthCappedMergeScheduler}. */
+public class TestBandwidthCappedMergeScheduler extends LuceneTestCase {
+
+ public void testBasicFunctionality() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler(100.0);
+
+ // Test initial value
+ assertEquals(100.0, scheduler.getMaxMbPerSec(), 0.001);
Review Comment:
Do we need this delta? Is `assertEquals` not working without it?
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "nipunbatra8 (via GitHub)" <gi...@apache.org>.
nipunbatra8 commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2308577087
##########
lucene/sandbox/src/java/org/apache/lucene/sandbox/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+import org.apache.lucene.index.MergeScheduler;
+
+/**
+ * A {@link MergeScheduler} that caps total IO write bandwidth across all running merges to a
+ * specified max MB/sec bandwidth.
+ *
+ * @lucene.experimental
+ */
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Global bandwidth cap in MB/s. Mutable so that updates are applied live. */
+ private double bandwidthMbPerSec;
+
+ /** Create a scheduler with a required global bandwidth cap in MB/s. */
+ public BandwidthCappedMergeScheduler(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
+ || Double.isNaN(bandwidthMbPerSec)
+ || Double.isInfinite(bandwidthMbPerSec)) {
+ throw new IllegalArgumentException("bandwidthMbPerSec must be a finite positive value");
+ }
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ }
+
+ /**
+ * Auto IO throttling is managed by this scheduler's bandwidth cap. Enabling parent CMS IO
+ * throttling is ignored.
+ */
+ @Override
+ public synchronized void enableAutoIOThrottle() {
+ if (verbose()) {
+ message("Ignoring enableAutoIOThrottle; using bandwidth cap instead");
+ }
+ // Intentionally no-op
+ }
+
+ /** Ensure auto IO throttling remains disabled. */
+ @Override
+ public synchronized void disableAutoIOThrottle() {
+ // Make sure parent state is disabled if it was somehow enabled earlier
+ super.disableAutoIOThrottle();
+ }
+
+ /** Always returns false since CMS auto IO throttling is disabled for this scheduler. */
+ @Override
+ public synchronized boolean getAutoIOThrottle() {
+ return false;
+ }
+
+ /** Get the global bandwidth cap in MB/s */
+ public double getMaxMbPerSec() {
+ return bandwidthMbPerSec;
+ }
+
+ /**
+ * Set the global bandwidth cap in MB/s.
+ *
+ * <p>This setting is live: merges that are already running will be adjusted to the new per-merge
+ * rate derived from this cap, without requiring a restart.
+ *
+ * @param bandwidthMbPerSec the new global bandwidth cap in MB/s; must be finite and > 0
+ */
+ public synchronized void setMaxMbPerSec(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
+ || Double.isNaN(bandwidthMbPerSec)
+ || Double.isInfinite(bandwidthMbPerSec)) {
+ throw new IllegalArgumentException("bandwidthMbPerSec must be a finite positive value");
+ }
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+
+ // Use the effective max thread count to avoid counting threads that CMS has paused
+ int effectiveMaxThreads = getMaxThreadCount();
+ if (effectiveMaxThreads == ConcurrentMergeScheduler.AUTO_DETECT_MERGES_AND_THREADS) {
+ int coreCount = Runtime.getRuntime().availableProcessors();
+ effectiveMaxThreads = Math.max(1, coreCount / 2);
+ }
+ int divisor = Math.min(effectiveMaxThreads, activeMerges);
+
+ double perMergeRate;
+ if (divisor > 0) {
+ perMergeRate = Math.max(0.1, bandwidthMbPerSec / divisor); // Use 0.1 MB/s minimum
+ } else {
+ perMergeRate = Double.POSITIVE_INFINITY;
+ }
+
+ // Apply the calculated rate limit to each active merge thread without unpausing paused threads
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ double currentRate = mergeThread.getRateLimiter().getMBPerSec();
+ if (currentRate > 0.0) { // Only update if not paused by parent CMS (above soft limit)
+ mergeThread.getRateLimiter().setMBPerSec(perMergeRate);
+ }
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthMbPerSec="
+ + bandwidthMbPerSec
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ }
+
+ @Override
+ public void run() {
+ long startTimeNS = System.nanoTime();
+ OneMerge merge = getMerge();
+ try {
+ if (verbose()) {
+ message(
+ "Starting bandwidth-capped merge: (estimatedMergeMB="
+ + merge.estimatedMergeBytes / (1024.0 * 1024.0)
+ + " MB)");
+ }
+ super.run(); // IO throttling is handled by the RateLimiter
+ } finally {
+ long durationNS = System.nanoTime() - startTimeNS;
+ if (verbose()) {
+ double durationMS = durationNS / 1_000_000.0;
+ double mbPerSec = merge.estimatedMergeBytes / (1024.0 * 1024.0) / (durationMS / 1000.0);
+ message(
+ "Merge completed: "
+ + merge.estimatedMergeBytes / (1024.0 * 1024.0)
+ + " MB in "
+ + String.format(java.util.Locale.US, "%.1f", durationMS)
Review Comment:
No, will make it more universal with Locale.ROOT
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "ytgu (via GitHub)" <gi...@apache.org>.
ytgu commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2317243178
##########
lucene/sandbox/src/java/org/apache/lucene/sandbox/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+import org.apache.lucene.index.MergeScheduler;
+
+/**
+ * A {@link MergeScheduler} that caps total IO write bandwidth across all running merges to a
+ * specified max MB/sec bandwidth.
+ *
+ * @lucene.experimental
+ */
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Global bandwidth cap in MB/s. Mutable so that updates are applied live. */
+ private double bandwidthMbPerSec;
+
+ /** Create a scheduler with a required global bandwidth cap in MB/s. */
+ public BandwidthCappedMergeScheduler(double bandwidthMbPerSec) {
+ validateBandwidth(bandwidthMbPerSec);
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ }
+
+ /** Validates that the bandwidth value is finite and positive. */
+ private static void validateBandwidth(double bandwidthMbPerSec) {
+ if (bandwidthMbPerSec <= 0.0
+ || Double.isNaN(bandwidthMbPerSec)
+ || Double.isInfinite(bandwidthMbPerSec)) {
+ throw new IllegalArgumentException("bandwidthMbPerSec must be a finite positive value");
+ }
+ }
+
+ /**
+ * Auto IO throttling is managed by this scheduler's bandwidth cap. Enabling parent CMS IO
+ * throttling is ignored.
+ */
+ @Override
+ public synchronized void enableAutoIOThrottle() {
+ if (verbose()) {
+ message("Ignoring enableAutoIOThrottle; using bandwidth cap instead");
+ }
+ // Intentionally no-op
+ }
+
+ /** Get the global bandwidth cap in MB/s */
+ public double getMaxMbPerSec() {
+ return bandwidthMbPerSec;
+ }
+
+ /**
+ * Set the global bandwidth cap in MB/s.
+ *
+ * <p>This setting is live: merges that are already running will be adjusted to the new per-merge
+ * rate derived from this cap, without requiring a restart.
+ *
+ * @param bandwidthMbPerSec the new global bandwidth cap in MB/s; must be finite and > 0
+ */
+ public synchronized void setMaxMbPerSec(double bandwidthMbPerSec) {
+ validateBandwidth(bandwidthMbPerSec);
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+
+ // Use the effective max thread count to avoid counting threads that CMS has paused
+ int effectiveMaxThreads = getMaxThreadCount();
+ if (effectiveMaxThreads == ConcurrentMergeScheduler.AUTO_DETECT_MERGES_AND_THREADS) {
+ int coreCount = Runtime.getRuntime().availableProcessors();
+ effectiveMaxThreads = Math.max(1, coreCount / 2);
+ }
+ int divisor = Math.min(effectiveMaxThreads, activeMerges);
Review Comment:
I saw your discussion with @mikemccand about using `min(maxThreadCount, activeMerges)` here to make sure the paused (even though they're alive) threads are not being counted. Just curious - can't we just use
```
for (MergeThread mergeThread : mergeThreads) {
if (mergeThread.isAlive() && mergeThread.getRateLimiter().getMBPerSec() > 0.0) {
activeMerges++;
}
}
```
to count the active merging threads?
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "shubhamsrkdev (via GitHub)" <gi...@apache.org>.
shubhamsrkdev commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2244963804
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
Review Comment:
nit: Bracket indentation.
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "stefanvodita (via GitHub)" <gi...@apache.org>.
stefanvodita commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2245001419
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
+ : Double.POSITIVE_INFINITY;
+
+ // Apply the calculated rate limit to each active merge thread
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ mergeThread.rateLimiter.setMBPerSec(perMergeRate);
Review Comment:
I'm curious how this behaves in performance tests. I'm guessing work isn't generally evenly distributed among threads.
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
Review Comment:
It will be interesting to see how we tweak these constants based on the tests we do. Would it make sense to make them configurable?
##########
lucene/core/src/test/org/apache/lucene/index/TestBandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,706 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import com.carrotsearch.randomizedtesting.generators.RandomStrings;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.KnnFloatVectorField;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.document.TextField;
+import org.apache.lucene.index.IndexWriterConfig.OpenMode;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.tests.analysis.MockAnalyzer;
+import org.apache.lucene.tests.store.MockDirectoryWrapper;
+import org.apache.lucene.tests.util.LuceneTestCase;
+import org.apache.lucene.tests.util.TestUtil;
+import org.apache.lucene.util.InfoStream;
+import org.apache.lucene.util.SuppressForbidden;
+
+/** Comprehensive tests for {@link BandwidthCappedMergeScheduler}. */
+public class TestBandwidthCappedMergeScheduler extends LuceneTestCase {
+
+ public void testBasicFunctionality() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+
+ // Test default values
+ assertEquals(1000.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testBandwidthRateBucketConfiguration() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+
+ // Test setting valid bandwidth rates
+ scheduler.setBandwidthRateBucket(100.0);
+ assertEquals(100.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.setBandwidthRateBucket(5.0); // minimum
+ assertEquals(5.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.setBandwidthRateBucket(10240.0); // maximum
+ assertEquals(10240.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testInvalidBandwidthRateConfiguration() throws Exception {
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+
+ // Test invalid bandwidth rates
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(4.0); // below minimum
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(10241.0); // above maximum
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(0.0);
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(-1.0);
+ });
+
+ scheduler.close();
+ }
+
+ public void testToString() throws Exception {
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ String str = scheduler.toString();
+
+ assertTrue(str.contains("BandwidthCappedMergeScheduler"));
+ assertTrue(str.contains("bandwidthRateBucket"));
+ assertTrue(str.contains("MB/s"));
+
+ scheduler.setBandwidthRateBucket(50.0);
+ str = scheduler.toString();
+ assertTrue(str.contains("50.0"));
+
+ scheduler.close();
+ }
+
+ public void testWithIndexWriter() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(100.0); // Set reasonable bandwidth
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2); // Force frequent flushes
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ // Add some documents to potentially trigger merges
+ for (int i = 0; i < 20; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(
+ new TextField(
+ "content",
+ "content " + i + " " + RandomStrings.randomRealisticUnicodeOfLength(random(), 100),
+ Field.Store.YES));
+ writer.addDocument(doc);
+
+ if (i % 5 == 0) {
+ writer.commit(); // Force segments
+ }
+ }
+
+ writer.forceMerge(1); // This should trigger merges
+ }
+
+ // The scheduler should have been used
+ assertTrue("Scheduler should have been initialized", scheduler.getBandwidthRateBucket() > 0);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testBandwidthDistributionAmongMerges() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(100.0); // 100 MB/s bucket
+ scheduler.setMaxMergesAndThreads(3, 2); // Allow multiple concurrent merges
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2);
+
+ // Use a merge policy that creates more merges
+ LogDocMergePolicy mp = new LogDocMergePolicy();
+ mp.setMergeFactor(2);
+ config.setMergePolicy(mp);
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ // Add many documents to trigger multiple merges
+ for (int i = 0; i < 50; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(
+ new TextField(
+ "content",
+ RandomStrings.randomRealisticUnicodeOfLength(random(), 200),
+ Field.Store.YES));
+ writer.addDocument(doc);
+
+ if (i % 10 == 0) {
+ writer.commit();
+ }
+ }
+
+ writer.forceMerge(1);
+ }
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testNoExtraFiles() throws IOException {
+ Directory directory = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(50.0);
+
+ IndexWriterConfig config = newIndexWriterConfig(new MockAnalyzer(random()));
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2);
+
+ IndexWriter writer = new IndexWriter(directory, config);
+
+ for (int iter = 0; iter < 5; iter++) {
+ if (VERBOSE) {
+ System.out.println("TEST: iter=" + iter);
+ }
+
+ for (int j = 0; j < 15; j++) {
+ Document doc = new Document();
+ doc.add(
+ newTextField(
+ "content",
+ "a b c " + RandomStrings.randomRealisticUnicodeOfLength(random(), 50),
+ Field.Store.NO));
+ writer.addDocument(doc);
+ }
+
+ writer.close();
+ TestIndexWriter.assertNoUnreferencedFiles(directory, "testNoExtraFiles");
+
+ // Reopen
+ config = newIndexWriterConfig(new MockAnalyzer(random()));
+ config.setMergeScheduler(new BandwidthCappedMergeScheduler());
+ config.setOpenMode(OpenMode.APPEND);
+ config.setMaxBufferedDocs(2);
+ writer = new IndexWriter(directory, config);
+ }
+
+ writer.close();
+ directory.close();
+ }
+
+ public void testDeleteMerging() throws IOException {
+ Directory directory = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(100.0);
+
+ LogDocMergePolicy mp = new LogDocMergePolicy();
+ mp.setMinMergeDocs(1000);
+
+ IndexWriterConfig config = newIndexWriterConfig(new MockAnalyzer(random()));
+ config.setMergeScheduler(scheduler);
+ config.setMergePolicy(mp);
+
+ IndexWriter writer = new IndexWriter(directory, config);
+ TestUtil.reduceOpenFiles(writer);
+
+ Document doc = new Document();
+ Field idField = newStringField("id", "", Field.Store.YES);
+ doc.add(idField);
+
+ for (int i = 0; i < 5; i++) {
+ if (VERBOSE) {
+ System.out.println("\nTEST: cycle " + i);
+ }
+ for (int j = 0; j < 50; j++) {
+ idField.setStringValue(Integer.toString(i * 50 + j));
+ writer.addDocument(doc);
+ }
+
+ int delID = i;
+ while (delID < 50 * (1 + i)) {
+ if (VERBOSE) {
+ System.out.println("TEST: del " + delID);
+ }
+ writer.deleteDocuments(new Term("id", "" + delID));
+ delID += 5;
+ }
+
+ writer.commit();
+ }
+
+ writer.close();
+ IndexReader reader = DirectoryReader.open(directory);
+ // Verify that we did not lose any deletes
+ // We add 5 cycles * 50 docs = 250 docs total
+ // We delete 5 cycles * 10 deletes per cycle = 50 deletes total
+ // So we should have 250 - 50 = 200 docs, but the actual behavior may vary
+ // Let's just verify we have a reasonable number of docs
+ assertTrue("Should have some docs remaining after deletes", reader.numDocs() > 0);
+ assertTrue("Should have fewer docs than originally added", reader.numDocs() < 250);
+ reader.close();
+ directory.close();
+ }
+
+ @SuppressForbidden(reason = "Thread sleep")
+ public void testMergeThreadTracking() throws Exception {
+ Directory dir = newDirectory();
+ Set<Thread> mergeThreadSet = ConcurrentHashMap.newKeySet();
+
+ // Track merge threads
+ BandwidthCappedMergeScheduler trackingScheduler =
+ new BandwidthCappedMergeScheduler() {
+ @Override
+ protected synchronized MergeThread getMergeThread(
+ MergeSource mergeSource, MergePolicy.OneMerge merge) throws IOException {
+ MergeThread thread = super.getMergeThread(mergeSource, merge);
+ mergeThreadSet.add(thread);
+ return thread;
+ }
+ };
+ trackingScheduler.setBandwidthRateBucket(50.0);
+
+ IndexWriterConfig iwc = new IndexWriterConfig(new MockAnalyzer(random()));
+ iwc.setMergeScheduler(trackingScheduler);
+ iwc.setMaxBufferedDocs(2);
+
+ LogMergePolicy lmp = newLogMergePolicy();
+ lmp.setMergeFactor(2);
+ iwc.setMergePolicy(lmp);
+
+ IndexWriter w = new IndexWriter(dir, iwc);
+ Document doc = new Document();
+ doc.add(new TextField("foo", "content", Field.Store.NO));
+
+ for (int i = 0; i < 10; i++) {
+ w.addDocument(doc);
+ }
+
+ w.close();
+
+ // Wait for merge threads to complete
+ for (Thread t : mergeThreadSet) {
+ t.join(5000); // Wait up to 5 seconds
+ }
+
+ dir.close();
+ }
+
+ public void testInfoStreamOutput() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(50.0);
+
+ List<String> messages = Collections.synchronizedList(new ArrayList<>());
+ InfoStream infoStream =
+ new InfoStream() {
+ @Override
+ public void close() {}
+
+ @Override
+ public void message(String component, String message) {
+ if (component.equals("MS")) {
+ messages.add(message);
+ }
+ }
+
+ @Override
+ public boolean isEnabled(String component) {
+ return component.equals("MS");
+ }
+ };
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setInfoStream(infoStream);
+ config.setMaxBufferedDocs(2);
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ for (int i = 0; i < 8; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(new TextField("content", "content " + i, Field.Store.YES));
+ writer.addDocument(doc);
+ }
+
+ writer.commit();
+ }
+
+ // Should have some merge-related messages
+ assertFalse("Should have some info stream messages", messages.isEmpty());
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testExceptionHandlingDuringMerge() throws IOException {
+ MockDirectoryWrapper directory = newMockDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(50.0);
+
+ IndexWriterConfig iwc = newIndexWriterConfig(new MockAnalyzer(random()));
+ iwc.setMergeScheduler(scheduler);
+ iwc.setMaxBufferedDocs(2);
+
+ IndexWriter writer = new IndexWriter(directory, iwc);
+ Document doc = new Document();
+ Field idField = newStringField("id", "", Field.Store.YES);
+ doc.add(idField);
+
+ // Add documents
+ for (int i = 0; i < 10; i++) {
+ idField.setStringValue(Integer.toString(i));
+ writer.addDocument(doc);
+ }
+
+ try {
+ writer.close();
+ } catch (Exception e) {
+ // Expected - some exceptions might occur during close
+ if (VERBOSE) {
+ System.out.println("Exception during close (expected): " + e.getMessage());
+ }
+ }
+
+ directory.close();
+ }
+
+ public void testMergeWithKnnVectors() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(100.0);
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2);
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ for (int i = 0; i < 10; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(
+ new KnnFloatVectorField(
+ "vector", new float[] {random().nextFloat(), random().nextFloat()}));
+ writer.addDocument(doc);
+ }
+
+ writer.forceMerge(1);
+ assertEquals(1, writer.getSegmentCount());
+ }
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testLargeBandwidthBucket() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(5000.0); // 5 GB/s
+
+ assertEquals(5000.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2);
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ for (int i = 0; i < 20; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(
+ new TextField(
+ "content",
+ RandomStrings.randomRealisticUnicodeOfLength(random(), 500),
+ Field.Store.YES));
+ writer.addDocument(doc);
+ }
+
+ writer.forceMerge(1);
+ }
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testThreadSafety() throws IOException, InterruptedException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(100.0);
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2);
+
+ final AtomicBoolean failed = new AtomicBoolean(false);
+ final CountDownLatch startLatch = new CountDownLatch(1);
+ final CountDownLatch doneLatch = new CountDownLatch(3);
+
+ // Create multiple threads that access bandwidth methods concurrently
+ for (int i = 0; i < 3; i++) {
+ Thread t =
+ new Thread(
+ () -> {
+ try {
+ startLatch.await();
+
+ for (int j = 0; j < 100; j++) {
+ double bandwidth = scheduler.getBandwidthRateBucket();
+ assertTrue("Bandwidth should be positive", bandwidth > 0);
+
+ // Test setting bandwidth
+ scheduler.setBandwidthRateBucket(50.0 + (j % 10));
+
+ Thread.yield();
+ }
+ } catch (Exception e) {
+ failed.set(true);
+ e.printStackTrace();
+ } finally {
+ doneLatch.countDown();
+ }
+ });
+ t.start();
+ }
+
+ startLatch.countDown();
+ assertTrue("Threads should complete within timeout", doneLatch.await(10, TimeUnit.SECONDS));
+ assertFalse("No thread should have failed", failed.get());
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testMergeSchedulerInheritance() throws IOException {
Review Comment:
I appreciate that we're being thorough, but I wonder if we risk distracting from more valuable tests with tests like this one. Can we go through this file and reconsider if we want to include all these tests? I think we can cut a lot.
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
+ : Double.POSITIVE_INFINITY;
+
+ // Apply the calculated rate limit to each active merge thread
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ mergeThread.rateLimiter.setMBPerSec(perMergeRate);
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthRateBucket="
+ + bandwidthRateBucket
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+ private final double mergeBandwidthMB;
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ this.mergeBandwidthMB = merge.estimatedMergeBytes / (1024.0 * 1024.0);
+ }
+
+ @Override
+ public void run() {
+ long startTime = System.currentTimeMillis();
+ try {
+ if (verbose()) {
+ message(
+ "Starting bandwidth-capped merge: "
+ + getSegmentName(merge)
+ + " (estimated="
+ + mergeBandwidthMB
+ + " MB)");
+ }
+ super.run(); // IO throttling is handled by the RateLimiter
+ } finally {
+ long duration = System.currentTimeMillis() - startTime;
+ if (verbose()) {
+ double mbPerSec = mergeBandwidthMB / Math.max(duration / 1000.0, 0.001);
Review Comment:
The way we compute this looks a bit funny. Do we need it?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
Review Comment:
Should the name have the unit of measurement?
##########
lucene/core/src/test/org/apache/lucene/index/TestBandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,706 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import com.carrotsearch.randomizedtesting.generators.RandomStrings;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.KnnFloatVectorField;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.document.TextField;
+import org.apache.lucene.index.IndexWriterConfig.OpenMode;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.tests.analysis.MockAnalyzer;
+import org.apache.lucene.tests.store.MockDirectoryWrapper;
+import org.apache.lucene.tests.util.LuceneTestCase;
+import org.apache.lucene.tests.util.TestUtil;
+import org.apache.lucene.util.InfoStream;
+import org.apache.lucene.util.SuppressForbidden;
+
+/** Comprehensive tests for {@link BandwidthCappedMergeScheduler}. */
+public class TestBandwidthCappedMergeScheduler extends LuceneTestCase {
+
+ public void testBasicFunctionality() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+
+ // Test default values
+ assertEquals(1000.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testBandwidthRateBucketConfiguration() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+
+ // Test setting valid bandwidth rates
+ scheduler.setBandwidthRateBucket(100.0);
+ assertEquals(100.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.setBandwidthRateBucket(5.0); // minimum
+ assertEquals(5.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.setBandwidthRateBucket(10240.0); // maximum
+ assertEquals(10240.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testInvalidBandwidthRateConfiguration() throws Exception {
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+
+ // Test invalid bandwidth rates
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(4.0); // below minimum
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(10241.0); // above maximum
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(0.0);
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(-1.0);
+ });
+
+ scheduler.close();
+ }
+
+ public void testToString() throws Exception {
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ String str = scheduler.toString();
+
+ assertTrue(str.contains("BandwidthCappedMergeScheduler"));
+ assertTrue(str.contains("bandwidthRateBucket"));
+ assertTrue(str.contains("MB/s"));
+
+ scheduler.setBandwidthRateBucket(50.0);
+ str = scheduler.toString();
+ assertTrue(str.contains("50.0"));
+
+ scheduler.close();
+ }
+
+ public void testWithIndexWriter() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(100.0); // Set reasonable bandwidth
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2); // Force frequent flushes
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ // Add some documents to potentially trigger merges
+ for (int i = 0; i < 20; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(
+ new TextField(
+ "content",
+ "content " + i + " " + RandomStrings.randomRealisticUnicodeOfLength(random(), 100),
+ Field.Store.YES));
+ writer.addDocument(doc);
+
+ if (i % 5 == 0) {
+ writer.commit(); // Force segments
+ }
+ }
+
+ writer.forceMerge(1); // This should trigger merges
+ }
+
+ // The scheduler should have been used
+ assertTrue("Scheduler should have been initialized", scheduler.getBandwidthRateBucket() > 0);
Review Comment:
Doesn't this just check line 125 without the lines in between having an impact?
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "nipunbatra8 (via GitHub)" <gi...@apache.org>.
nipunbatra8 commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2252161206
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
+ : Double.POSITIVE_INFINITY;
+
+ // Apply the calculated rate limit to each active merge thread
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ mergeThread.rateLimiter.setMBPerSec(perMergeRate);
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthRateBucket="
+ + bandwidthRateBucket
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+ private final double mergeBandwidthMB;
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ this.mergeBandwidthMB = merge.estimatedMergeBytes / (1024.0 * 1024.0);
+ }
+
+ @Override
+ public void run() {
+ long startTime = System.currentTimeMillis();
+ try {
+ if (verbose()) {
+ message(
+ "Starting bandwidth-capped merge: "
+ + getSegmentName(merge)
+ + " (estimated="
+ + mergeBandwidthMB
+ + " MB)");
+ }
+ super.run(); // IO throttling is handled by the RateLimiter
+ } finally {
+ long duration = System.currentTimeMillis() - startTime;
+ if (verbose()) {
+ double mbPerSec = mergeBandwidthMB / Math.max(duration / 1000.0, 0.001);
Review Comment:
The duration or mbPerSec? I have the 0.001 to avoid division by zero problems in case the duration was computed as 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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "nipunbatra8 (via GitHub)" <gi...@apache.org>.
nipunbatra8 commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2260905136
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
Review Comment:
Yes
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14964:
URL: https://github.com/apache/lucene/pull/14964#issuecomment-3408720336
This PR has not had activity in the past 2 weeks, labeling it as stale. If the PR is waiting for review, notify the dev@lucene.apache.org list. Thank you for your contribution!
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "nipunbatra8 (via GitHub)" <gi...@apache.org>.
nipunbatra8 commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2301672163
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
Review Comment:
Yes, made it configurable.
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "nipunbatra8 (via GitHub)" <gi...@apache.org>.
nipunbatra8 commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2308528094
##########
lucene/sandbox/src/java/org/apache/lucene/sandbox/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+import org.apache.lucene.index.MergeScheduler;
+
+/**
+ * A {@link MergeScheduler} that caps total IO write bandwidth across all running merges to a
+ * specified max MB/sec bandwidth.
+ *
+ * @lucene.experimental
+ */
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Global bandwidth cap in MB/s. Mutable so that updates are applied live. */
+ private double bandwidthMbPerSec;
+
+ /** Create a scheduler with a required global bandwidth cap in MB/s. */
+ public BandwidthCappedMergeScheduler(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
+ || Double.isNaN(bandwidthMbPerSec)
+ || Double.isInfinite(bandwidthMbPerSec)) {
+ throw new IllegalArgumentException("bandwidthMbPerSec must be a finite positive value");
+ }
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ }
+
+ /**
+ * Auto IO throttling is managed by this scheduler's bandwidth cap. Enabling parent CMS IO
+ * throttling is ignored.
+ */
+ @Override
+ public synchronized void enableAutoIOThrottle() {
+ if (verbose()) {
+ message("Ignoring enableAutoIOThrottle; using bandwidth cap instead");
+ }
+ // Intentionally no-op
+ }
+
+ /** Ensure auto IO throttling remains disabled. */
+ @Override
+ public synchronized void disableAutoIOThrottle() {
+ // Make sure parent state is disabled if it was somehow enabled earlier
Review Comment:
The parent class has a IOThrottle that can be enabled for more efficiency. Realized it is defaulted as false in the parent class so this method and the one below are not needed because it will always be false since we disabled the enableAutoIOThrottle function.
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "shubhamsrkdev (via GitHub)" <gi...@apache.org>.
shubhamsrkdev commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2306941801
##########
lucene/sandbox/src/java/org/apache/lucene/sandbox/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+import org.apache.lucene.index.MergeScheduler;
+
+/**
+ * A {@link MergeScheduler} that caps total IO write bandwidth across all running merges to a
+ * specified max MB/sec bandwidth.
+ *
+ * @lucene.experimental
+ */
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Global bandwidth cap in MB/s. Mutable so that updates are applied live. */
+ private double bandwidthMbPerSec;
+
+ /** Create a scheduler with a required global bandwidth cap in MB/s. */
+ public BandwidthCappedMergeScheduler(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
+ || Double.isNaN(bandwidthMbPerSec)
+ || Double.isInfinite(bandwidthMbPerSec)) {
+ throw new IllegalArgumentException("bandwidthMbPerSec must be a finite positive value");
+ }
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ }
+
+ /**
+ * Auto IO throttling is managed by this scheduler's bandwidth cap. Enabling parent CMS IO
+ * throttling is ignored.
+ */
+ @Override
+ public synchronized void enableAutoIOThrottle() {
+ if (verbose()) {
+ message("Ignoring enableAutoIOThrottle; using bandwidth cap instead");
+ }
+ // Intentionally no-op
+ }
+
+ /** Ensure auto IO throttling remains disabled. */
+ @Override
+ public synchronized void disableAutoIOThrottle() {
+ // Make sure parent state is disabled if it was somehow enabled earlier
Review Comment:
Maybe I am missing something here - What would be the scenario when/how this would be enabled?
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "mikemccand (via GitHub)" <gi...@apache.org>.
mikemccand commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2256819456
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
Review Comment:
Maybe state this more succinctly: `A {@link MergeScheduler} that caps total IO write bandwidth across all running merges to a specified max MB/sec bandwidth.`?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
Review Comment:
Maybe move this to Lucene's sandbox module? It's where we land more experimental thingies and give them a chance to be used / thrive / iterate and eventually graduate to core.
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
Review Comment:
Could you add to the javadoc that this setting is "live", meaning any running merges running now will be updated to the new cap too?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
Review Comment:
Could we expand this into true `if` (don't use the ternary operator)?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
+ : Double.POSITIVE_INFINITY;
+
+ // Apply the calculated rate limit to each active merge thread
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ mergeThread.rateLimiter.setMBPerSec(perMergeRate);
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthRateBucket="
+ + bandwidthRateBucket
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+ private final double mergeBandwidthMB;
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ this.mergeBandwidthMB = merge.estimatedMergeBytes / (1024.0 * 1024.0);
Review Comment:
Oh, this is not actually a bandwidth right? It's just the estimated final segment size, in MB?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
Review Comment:
I think this is implicitly done in java? Even if we add the starting `mbPerSec` to the ctor?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
+ : Double.POSITIVE_INFINITY;
+
+ // Apply the calculated rate limit to each active merge thread
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ mergeThread.rateLimiter.setMBPerSec(perMergeRate);
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthRateBucket="
+ + bandwidthRateBucket
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+ private final double mergeBandwidthMB;
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ this.mergeBandwidthMB = merge.estimatedMergeBytes / (1024.0 * 1024.0);
+ }
+
+ @Override
+ public void run() {
+ long startTime = System.currentTimeMillis();
+ try {
+ if (verbose()) {
+ message(
+ "Starting bandwidth-capped merge: "
+ + getSegmentName(merge)
+ + " (estimated="
+ + mergeBandwidthMB
+ + " MB)");
+ }
+ super.run(); // IO throttling is handled by the RateLimiter
+ } finally {
+ long duration = System.currentTimeMillis() - startTime;
+ if (verbose()) {
+ double mbPerSec = mergeBandwidthMB / Math.max(duration / 1000.0, 0.001);
+ message(
+ "Merge completed: "
+ + getSegmentName(merge)
+ + " "
+ + mergeBandwidthMB
+ + " MB in "
+ + duration
+ + "ms ("
+ + String.format(java.util.Locale.US, "%.2f", mbPerSec)
+ + " MB/s)");
+ }
+ }
+ }
+ }
+
+ private static String getSegmentName(OneMerge merge) {
+ return merge.info != null ? merge.info.info.name : "_na_";
Review Comment:
Hmm is `merge.info` sometimes `null`? It shouldn't ever be, I think?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
Review Comment:
Maybe don't talk about buckets? I'm not sure what they are in this context.
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
Review Comment:
Do we really need the `min` and `max`?
I.e. the min could be 1 [ULP](https://en.wikipedia.org/wiki/Unit_in_the_last_place) above 0.0, and the max is unbounded (`Double.MAX_VALUE`). In practice very large values of the cap won't matter because merges will become CPU bound at that point and cannot use so much bandwidth.
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
Review Comment:
Aha, thank you! This is because PR is still draft right?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
Review Comment:
`getMaxMbPerSec`?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
+ : Double.POSITIVE_INFINITY;
+
+ // Apply the calculated rate limit to each active merge thread
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ mergeThread.rateLimiter.setMBPerSec(perMergeRate);
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthRateBucket="
+ + bandwidthRateBucket
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+ private final double mergeBandwidthMB;
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ this.mergeBandwidthMB = merge.estimatedMergeBytes / (1024.0 * 1024.0);
+ }
+
+ @Override
+ public void run() {
+ long startTime = System.currentTimeMillis();
Review Comment:
Let's use `System.nanoTime()` instead? It is guaranteed monotonic, and has no relation to the odd human labeling of time (which we do not need since we are just computing duration).
Also let's add the units in the variable name (`startTimeNS`).
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
Review Comment:
Let's detect if `doAutioIOThrottle` is enabled and throw some exception somewhere? That CMS feature is incompatible with this class, I think, so it should be clear to the user. Oh, maybe simply override `enable/disableAutoIOThrottle`?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
+ : Double.POSITIVE_INFINITY;
+
+ // Apply the calculated rate limit to each active merge thread
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ mergeThread.rateLimiter.setMBPerSec(perMergeRate);
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthRateBucket="
+ + bandwidthRateBucket
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+ private final double mergeBandwidthMB;
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ this.mergeBandwidthMB = merge.estimatedMergeBytes / (1024.0 * 1024.0);
+ }
+
+ @Override
+ public void run() {
+ long startTime = System.currentTimeMillis();
+ try {
+ if (verbose()) {
+ message(
+ "Starting bandwidth-capped merge: "
+ + getSegmentName(merge)
+ + " (estimated="
Review Comment:
`(estimatedMbPerSec=` instead?
Also, this is just the value when this merge kicks off? So maybe change to `startingMbPerSec=`?
Edit: oh this is the estimated size of the final merged segment in MB? Maybe just change to `merge.estimatedMergeBytes/1024./1024.` and no need for member variable?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
Review Comment:
Let's include units in the method name, e.g. `setMaxMbPerSec`?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
Review Comment:
`activeMerges` isn't right because it can include merge threads that are created and alive (in the JVM sense) but are paused by CMS parent (in the "soft merge limit" case). I think it suffices to take `min(maxThreadCount, activeMerges)` instead. But you'll need to make `maxThreadCount` member protected, or use `getMaxThreadCount()` instead, but also take steps to protect against the `AUTO_DETECT_MERGES_AND_THREADS` in case it didn't get default'd yet.
CMS has a soft limit (`maxThreadCount`) and a hard limit (`maxMergeCount`) on the number of merges requested by the `MergePolicy` that it will tolerate. Below or equal to the soft limit, everything is normal. When more merges are needed than the soft limit, the largest N merges are paused so that the remaining soft limit merges can run. If more merges than the hard limit are requested, CMS will apply backpressure by hijacking the incoming indexing threads (those threads creating new segments and requesting new merges), pausing them, until the merging can catch up and the number of merges requested is again <= the hard limit.
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
+ : Double.POSITIVE_INFINITY;
+
+ // Apply the calculated rate limit to each active merge thread
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ mergeThread.rateLimiter.setMBPerSec(perMergeRate);
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthRateBucket="
+ + bandwidthRateBucket
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+ private final double mergeBandwidthMB;
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ this.mergeBandwidthMB = merge.estimatedMergeBytes / (1024.0 * 1024.0);
+ }
+
+ @Override
+ public void run() {
+ long startTime = System.currentTimeMillis();
+ try {
+ if (verbose()) {
+ message(
+ "Starting bandwidth-capped merge: "
+ + getSegmentName(merge)
+ + " (estimated="
+ + mergeBandwidthMB
+ + " MB)");
+ }
+ super.run(); // IO throttling is handled by the RateLimiter
+ } finally {
+ long duration = System.currentTimeMillis() - startTime;
Review Comment:
`durationNS`?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
+ : Double.POSITIVE_INFINITY;
+
+ // Apply the calculated rate limit to each active merge thread
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ mergeThread.rateLimiter.setMBPerSec(perMergeRate);
+ }
+ }
+ }
+
+ /** Creates a custom merge thread with bandwidth tracking capabilities. */
+ @Override
+ protected synchronized MergeThread getMergeThread(MergeSource mergeSource, OneMerge merge)
+ throws IOException {
+ return new BandwidthTrackingMergeThread(mergeSource, merge);
+ }
+
+ /** Returns a string representation including the current bandwidth rate bucket setting. */
+ @Override
+ public String toString() {
+ return getClass().getSimpleName()
+ + ": "
+ + super.toString()
+ + ", bandwidthRateBucket="
+ + bandwidthRateBucket
+ + " MB/s";
+ }
+
+ /** Merge thread that logs the rate limiter value after merge completes. */
+ protected class BandwidthTrackingMergeThread extends MergeThread {
+ private final double mergeBandwidthMB;
+
+ /** Creates a new BandwidthTrackingMergeThread for the given merge. */
+ public BandwidthTrackingMergeThread(MergeSource mergeSource, OneMerge merge) {
+ super(mergeSource, merge);
+ this.mergeBandwidthMB = merge.estimatedMergeBytes / (1024.0 * 1024.0);
+ }
+
+ @Override
+ public void run() {
+ long startTime = System.currentTimeMillis();
+ try {
+ if (verbose()) {
+ message(
+ "Starting bandwidth-capped merge: "
+ + getSegmentName(merge)
+ + " (estimated="
+ + mergeBandwidthMB
+ + " MB)");
+ }
+ super.run(); // IO throttling is handled by the RateLimiter
+ } finally {
+ long duration = System.currentTimeMillis() - startTime;
+ if (verbose()) {
+ double mbPerSec = mergeBandwidthMB / Math.max(duration / 1000.0, 0.001);
Review Comment:
It's OK to just let the math produce `Double.POSITIVE_INFINITY`? It's only `int`/`long` etc. that cannot divide-by-zero?
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
+ throw new IllegalArgumentException(
+ "Bandwidth rate must be between "
+ + MIN_MERGE_MB_PER_SEC
+ + " and "
+ + MAX_MERGE_MB_PER_SEC
+ + " MB/s");
+ }
+ this.bandwidthRateBucket = mbPerSec;
+ updateMergeThreads();
+ }
+
+ /** Get the global bandwidth rate bucket in MB/s */
+ public double getBandwidthRateBucket() {
+ return bandwidthRateBucket;
+ }
+
+ /** Distributes the global bandwidth rate bucket evenly among all active merge threads. */
+ @Override
+ protected synchronized void updateMergeThreads() {
+ super.updateMergeThreads();
+ int activeMerges = 0;
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ activeMerges++;
+ }
+ }
+ double perMergeRate =
+ activeMerges > 0
+ ? Math.max(
+ MIN_MERGE_MB_PER_SEC,
+ Math.min(MAX_MERGE_MB_PER_SEC, bandwidthRateBucket / activeMerges))
+ : Double.POSITIVE_INFINITY;
+
+ // Apply the calculated rate limit to each active merge thread
+ for (MergeThread mergeThread : mergeThreads) {
+ if (mergeThread.isAlive()) {
+ mergeThread.rateLimiter.setMBPerSec(perMergeRate);
Review Comment:
This may indeed prove too simplistic (statically dividing bandwidth into the N running merges), but it's good for starters / MLP ("minimum lovable PR") to unlock fun testing.
Actually, instead of statically dividing the cap across N rate limiters ... couldn't we create a single `MergeRateLimiter` shared across all merge threads, set to the cap? This way, if one merge is doing CPU heavy part of merging and not able to use much IO bandwidth, other merges could surge over their 1/Nth static share? But this is kinda a big change... CMS seems to quietly assume/expect that each merge thread has its own rate limiter. It might freak out if you use the same rate limiter across all threads.
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
Review Comment:
Hmm why do we even have the `START_MB_PER_SEC`? Could we rather make the ctor take a required `mbPerSec` and use that as the starting value?
Does this class support updating the cap even while merges are running? It looks like it does, yay! (Maybe apps have time-dependent bandwidth availability).
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "mikemccand (via GitHub)" <gi...@apache.org>.
mikemccand commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2256808524
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
Review Comment:
+1 to include `MB_PER_SEC` in whichever name we settle on (maybe `bandwidthMbPerSec`?). Too many times I've seen "off by 1000" ish bugs happen because of units confusion ...
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "nipunbatra8 (via GitHub)" <gi...@apache.org>.
nipunbatra8 commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2252122671
##########
lucene/core/src/test/org/apache/lucene/index/TestBandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,706 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import com.carrotsearch.randomizedtesting.generators.RandomStrings;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.KnnFloatVectorField;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.document.TextField;
+import org.apache.lucene.index.IndexWriterConfig.OpenMode;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.tests.analysis.MockAnalyzer;
+import org.apache.lucene.tests.store.MockDirectoryWrapper;
+import org.apache.lucene.tests.util.LuceneTestCase;
+import org.apache.lucene.tests.util.TestUtil;
+import org.apache.lucene.util.InfoStream;
+import org.apache.lucene.util.SuppressForbidden;
+
+/** Comprehensive tests for {@link BandwidthCappedMergeScheduler}. */
+public class TestBandwidthCappedMergeScheduler extends LuceneTestCase {
+
+ public void testBasicFunctionality() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+
+ // Test default values
+ assertEquals(1000.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testBandwidthRateBucketConfiguration() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+
+ // Test setting valid bandwidth rates
+ scheduler.setBandwidthRateBucket(100.0);
+ assertEquals(100.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.setBandwidthRateBucket(5.0); // minimum
+ assertEquals(5.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.setBandwidthRateBucket(10240.0); // maximum
+ assertEquals(10240.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testInvalidBandwidthRateConfiguration() throws Exception {
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+
+ // Test invalid bandwidth rates
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(4.0); // below minimum
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(10241.0); // above maximum
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(0.0);
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setBandwidthRateBucket(-1.0);
+ });
+
+ scheduler.close();
+ }
+
+ public void testToString() throws Exception {
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ String str = scheduler.toString();
+
+ assertTrue(str.contains("BandwidthCappedMergeScheduler"));
+ assertTrue(str.contains("bandwidthRateBucket"));
+ assertTrue(str.contains("MB/s"));
+
+ scheduler.setBandwidthRateBucket(50.0);
+ str = scheduler.toString();
+ assertTrue(str.contains("50.0"));
+
+ scheduler.close();
+ }
+
+ public void testWithIndexWriter() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(100.0); // Set reasonable bandwidth
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2); // Force frequent flushes
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ // Add some documents to potentially trigger merges
+ for (int i = 0; i < 20; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(
+ new TextField(
+ "content",
+ "content " + i + " " + RandomStrings.randomRealisticUnicodeOfLength(random(), 100),
+ Field.Store.YES));
+ writer.addDocument(doc);
+
+ if (i % 5 == 0) {
+ writer.commit(); // Force segments
+ }
+ }
+
+ writer.forceMerge(1); // This should trigger merges
+ }
+
+ // The scheduler should have been used
+ assertTrue("Scheduler should have been initialized", scheduler.getBandwidthRateBucket() > 0);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testBandwidthDistributionAmongMerges() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(100.0); // 100 MB/s bucket
+ scheduler.setMaxMergesAndThreads(3, 2); // Allow multiple concurrent merges
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2);
+
+ // Use a merge policy that creates more merges
+ LogDocMergePolicy mp = new LogDocMergePolicy();
+ mp.setMergeFactor(2);
+ config.setMergePolicy(mp);
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ // Add many documents to trigger multiple merges
+ for (int i = 0; i < 50; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(
+ new TextField(
+ "content",
+ RandomStrings.randomRealisticUnicodeOfLength(random(), 200),
+ Field.Store.YES));
+ writer.addDocument(doc);
+
+ if (i % 10 == 0) {
+ writer.commit();
+ }
+ }
+
+ writer.forceMerge(1);
+ }
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testNoExtraFiles() throws IOException {
+ Directory directory = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(50.0);
+
+ IndexWriterConfig config = newIndexWriterConfig(new MockAnalyzer(random()));
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2);
+
+ IndexWriter writer = new IndexWriter(directory, config);
+
+ for (int iter = 0; iter < 5; iter++) {
+ if (VERBOSE) {
+ System.out.println("TEST: iter=" + iter);
+ }
+
+ for (int j = 0; j < 15; j++) {
+ Document doc = new Document();
+ doc.add(
+ newTextField(
+ "content",
+ "a b c " + RandomStrings.randomRealisticUnicodeOfLength(random(), 50),
+ Field.Store.NO));
+ writer.addDocument(doc);
+ }
+
+ writer.close();
+ TestIndexWriter.assertNoUnreferencedFiles(directory, "testNoExtraFiles");
+
+ // Reopen
+ config = newIndexWriterConfig(new MockAnalyzer(random()));
+ config.setMergeScheduler(new BandwidthCappedMergeScheduler());
+ config.setOpenMode(OpenMode.APPEND);
+ config.setMaxBufferedDocs(2);
+ writer = new IndexWriter(directory, config);
+ }
+
+ writer.close();
+ directory.close();
+ }
+
+ public void testDeleteMerging() throws IOException {
+ Directory directory = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(100.0);
+
+ LogDocMergePolicy mp = new LogDocMergePolicy();
+ mp.setMinMergeDocs(1000);
+
+ IndexWriterConfig config = newIndexWriterConfig(new MockAnalyzer(random()));
+ config.setMergeScheduler(scheduler);
+ config.setMergePolicy(mp);
+
+ IndexWriter writer = new IndexWriter(directory, config);
+ TestUtil.reduceOpenFiles(writer);
+
+ Document doc = new Document();
+ Field idField = newStringField("id", "", Field.Store.YES);
+ doc.add(idField);
+
+ for (int i = 0; i < 5; i++) {
+ if (VERBOSE) {
+ System.out.println("\nTEST: cycle " + i);
+ }
+ for (int j = 0; j < 50; j++) {
+ idField.setStringValue(Integer.toString(i * 50 + j));
+ writer.addDocument(doc);
+ }
+
+ int delID = i;
+ while (delID < 50 * (1 + i)) {
+ if (VERBOSE) {
+ System.out.println("TEST: del " + delID);
+ }
+ writer.deleteDocuments(new Term("id", "" + delID));
+ delID += 5;
+ }
+
+ writer.commit();
+ }
+
+ writer.close();
+ IndexReader reader = DirectoryReader.open(directory);
+ // Verify that we did not lose any deletes
+ // We add 5 cycles * 50 docs = 250 docs total
+ // We delete 5 cycles * 10 deletes per cycle = 50 deletes total
+ // So we should have 250 - 50 = 200 docs, but the actual behavior may vary
+ // Let's just verify we have a reasonable number of docs
+ assertTrue("Should have some docs remaining after deletes", reader.numDocs() > 0);
+ assertTrue("Should have fewer docs than originally added", reader.numDocs() < 250);
+ reader.close();
+ directory.close();
+ }
+
+ @SuppressForbidden(reason = "Thread sleep")
+ public void testMergeThreadTracking() throws Exception {
+ Directory dir = newDirectory();
+ Set<Thread> mergeThreadSet = ConcurrentHashMap.newKeySet();
+
+ // Track merge threads
+ BandwidthCappedMergeScheduler trackingScheduler =
+ new BandwidthCappedMergeScheduler() {
+ @Override
+ protected synchronized MergeThread getMergeThread(
+ MergeSource mergeSource, MergePolicy.OneMerge merge) throws IOException {
+ MergeThread thread = super.getMergeThread(mergeSource, merge);
+ mergeThreadSet.add(thread);
+ return thread;
+ }
+ };
+ trackingScheduler.setBandwidthRateBucket(50.0);
+
+ IndexWriterConfig iwc = new IndexWriterConfig(new MockAnalyzer(random()));
+ iwc.setMergeScheduler(trackingScheduler);
+ iwc.setMaxBufferedDocs(2);
+
+ LogMergePolicy lmp = newLogMergePolicy();
+ lmp.setMergeFactor(2);
+ iwc.setMergePolicy(lmp);
+
+ IndexWriter w = new IndexWriter(dir, iwc);
+ Document doc = new Document();
+ doc.add(new TextField("foo", "content", Field.Store.NO));
+
+ for (int i = 0; i < 10; i++) {
+ w.addDocument(doc);
+ }
+
+ w.close();
+
+ // Wait for merge threads to complete
+ for (Thread t : mergeThreadSet) {
+ t.join(5000); // Wait up to 5 seconds
+ }
+
+ dir.close();
+ }
+
+ public void testInfoStreamOutput() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(50.0);
+
+ List<String> messages = Collections.synchronizedList(new ArrayList<>());
+ InfoStream infoStream =
+ new InfoStream() {
+ @Override
+ public void close() {}
+
+ @Override
+ public void message(String component, String message) {
+ if (component.equals("MS")) {
+ messages.add(message);
+ }
+ }
+
+ @Override
+ public boolean isEnabled(String component) {
+ return component.equals("MS");
+ }
+ };
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setInfoStream(infoStream);
+ config.setMaxBufferedDocs(2);
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ for (int i = 0; i < 8; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(new TextField("content", "content " + i, Field.Store.YES));
+ writer.addDocument(doc);
+ }
+
+ writer.commit();
+ }
+
+ // Should have some merge-related messages
+ assertFalse("Should have some info stream messages", messages.isEmpty());
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testExceptionHandlingDuringMerge() throws IOException {
+ MockDirectoryWrapper directory = newMockDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(50.0);
+
+ IndexWriterConfig iwc = newIndexWriterConfig(new MockAnalyzer(random()));
+ iwc.setMergeScheduler(scheduler);
+ iwc.setMaxBufferedDocs(2);
+
+ IndexWriter writer = new IndexWriter(directory, iwc);
+ Document doc = new Document();
+ Field idField = newStringField("id", "", Field.Store.YES);
+ doc.add(idField);
+
+ // Add documents
+ for (int i = 0; i < 10; i++) {
+ idField.setStringValue(Integer.toString(i));
+ writer.addDocument(doc);
+ }
+
+ try {
+ writer.close();
+ } catch (Exception e) {
+ // Expected - some exceptions might occur during close
+ if (VERBOSE) {
+ System.out.println("Exception during close (expected): " + e.getMessage());
+ }
+ }
+
+ directory.close();
+ }
+
+ public void testMergeWithKnnVectors() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(100.0);
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2);
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ for (int i = 0; i < 10; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(
+ new KnnFloatVectorField(
+ "vector", new float[] {random().nextFloat(), random().nextFloat()}));
+ writer.addDocument(doc);
+ }
+
+ writer.forceMerge(1);
+ assertEquals(1, writer.getSegmentCount());
+ }
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testLargeBandwidthBucket() throws IOException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(5000.0); // 5 GB/s
+
+ assertEquals(5000.0, scheduler.getBandwidthRateBucket(), 0.001);
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2);
+
+ try (IndexWriter writer = new IndexWriter(dir, config)) {
+ for (int i = 0; i < 20; i++) {
+ Document doc = new Document();
+ doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
+ doc.add(
+ new TextField(
+ "content",
+ RandomStrings.randomRealisticUnicodeOfLength(random(), 500),
+ Field.Store.YES));
+ writer.addDocument(doc);
+ }
+
+ writer.forceMerge(1);
+ }
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testThreadSafety() throws IOException, InterruptedException {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler();
+ scheduler.setBandwidthRateBucket(100.0);
+
+ IndexWriterConfig config = newIndexWriterConfig();
+ config.setMergeScheduler(scheduler);
+ config.setMaxBufferedDocs(2);
+
+ final AtomicBoolean failed = new AtomicBoolean(false);
+ final CountDownLatch startLatch = new CountDownLatch(1);
+ final CountDownLatch doneLatch = new CountDownLatch(3);
+
+ // Create multiple threads that access bandwidth methods concurrently
+ for (int i = 0; i < 3; i++) {
+ Thread t =
+ new Thread(
+ () -> {
+ try {
+ startLatch.await();
+
+ for (int j = 0; j < 100; j++) {
+ double bandwidth = scheduler.getBandwidthRateBucket();
+ assertTrue("Bandwidth should be positive", bandwidth > 0);
+
+ // Test setting bandwidth
+ scheduler.setBandwidthRateBucket(50.0 + (j % 10));
+
+ Thread.yield();
+ }
+ } catch (Exception e) {
+ failed.set(true);
+ e.printStackTrace();
+ } finally {
+ doneLatch.countDown();
+ }
+ });
+ t.start();
+ }
+
+ startLatch.countDown();
+ assertTrue("Threads should complete within timeout", doneLatch.await(10, TimeUnit.SECONDS));
+ assertFalse("No thread should have failed", failed.get());
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testMergeSchedulerInheritance() throws IOException {
Review Comment:
Yes, just wanted to make sure to be through in case but I agree. Will cut down on test cases.
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "nipunbatra8 (via GitHub)" <gi...@apache.org>.
nipunbatra8 commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2308567962
##########
lucene/sandbox/src/java/org/apache/lucene/sandbox/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+import org.apache.lucene.index.MergeScheduler;
+
+/**
+ * A {@link MergeScheduler} that caps total IO write bandwidth across all running merges to a
+ * specified max MB/sec bandwidth.
+ *
+ * @lucene.experimental
+ */
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Global bandwidth cap in MB/s. Mutable so that updates are applied live. */
+ private double bandwidthMbPerSec;
+
+ /** Create a scheduler with a required global bandwidth cap in MB/s. */
+ public BandwidthCappedMergeScheduler(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
+ || Double.isNaN(bandwidthMbPerSec)
+ || Double.isInfinite(bandwidthMbPerSec)) {
+ throw new IllegalArgumentException("bandwidthMbPerSec must be a finite positive value");
+ }
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ }
+
+ /**
+ * Auto IO throttling is managed by this scheduler's bandwidth cap. Enabling parent CMS IO
+ * throttling is ignored.
+ */
+ @Override
+ public synchronized void enableAutoIOThrottle() {
+ if (verbose()) {
+ message("Ignoring enableAutoIOThrottle; using bandwidth cap instead");
+ }
+ // Intentionally no-op
+ }
+
+ /** Ensure auto IO throttling remains disabled. */
+ @Override
+ public synchronized void disableAutoIOThrottle() {
+ // Make sure parent state is disabled if it was somehow enabled earlier
+ super.disableAutoIOThrottle();
+ }
+
+ /** Always returns false since CMS auto IO throttling is disabled for this scheduler. */
+ @Override
+ public synchronized boolean getAutoIOThrottle() {
+ return false;
+ }
+
+ /** Get the global bandwidth cap in MB/s */
+ public double getMaxMbPerSec() {
+ return bandwidthMbPerSec;
+ }
+
+ /**
+ * Set the global bandwidth cap in MB/s.
+ *
+ * <p>This setting is live: merges that are already running will be adjusted to the new per-merge
+ * rate derived from this cap, without requiring a restart.
+ *
+ * @param bandwidthMbPerSec the new global bandwidth cap in MB/s; must be finite and > 0
+ */
+ public synchronized void setMaxMbPerSec(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
Review Comment:
Good idea, makes it more readable too
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "mikemccand (via GitHub)" <gi...@apache.org>.
mikemccand commented on PR #14964:
URL: https://github.com/apache/lucene/pull/14964#issuecomment-3159879637
> I wonder if we're addressing the original issue. Maybe I need to spend more time reading though it, but my understanding is that `MergeRateLimiter`, which we use here, is meant to control the amount of resources we put into the merging process, while the idea we were considering was to limit the size of the output of merges. Please correct me if I got that wrong.
I like this simplicity of this approach, vs the other ideas on https://github.com/apache/lucene/issues/14148 ... but I do agree that having the MP and/or MS aware that they are operating in a resource-scarce environment might do better. E.g. TMP might delay picking merges if it's already spent its short-term IO budget, and be forced to allow more deletions to accumulate.
But with this simplistic approach, we might emergently see similar behavior. Throttling a lerge merge that is trying to reclaim deletes effectively stretches out the lifetime of the segments being merged, allowing more deletions to accumulate that would normally, and delaying the appearance of the newly merged segment so that MP/MS cannot pick it for merging until itself is done being merged. We need to test this approach to see...
@nipunbatra8 `RateLimiter` (as hairy as it is!!) might be overly simplistic for your usage here, especially if we pivot the approach to share a single rate limiter across all merge threads ... it does not implement the "[leaky bucket](https://en.wikipedia.org/wiki/Leaky_bucket)" model to smooth out bursty IO. This is another reason to not pivot to this new approach just yet ...
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "shubhamsrkdev (via GitHub)" <gi...@apache.org>.
shubhamsrkdev commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2307011526
##########
lucene/sandbox/src/test/org/apache/lucene/sandbox/index/TestBandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import com.carrotsearch.randomizedtesting.generators.RandomStrings;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.document.TextField;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.index.LogDocMergePolicy;
+import org.apache.lucene.index.MergePolicy;
+import org.apache.lucene.index.MergeScheduler;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.tests.util.LuceneTestCase;
+import org.apache.lucene.util.InfoStream;
+
+/** Tests for {@link BandwidthCappedMergeScheduler}. */
+public class TestBandwidthCappedMergeScheduler extends LuceneTestCase {
+
+ public void testBasicFunctionality() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler(100.0);
+
+ // Test initial value
+ assertEquals(100.0, scheduler.getMaxMbPerSec(), 0.001);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testBandwidthConfiguration() throws Exception {
+ Directory dir = newDirectory();
Review Comment:
I am not sure about the usefulness of checking setters and getters functionalities. Same goes for previous test.
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "nipunbatra8 (via GitHub)" <gi...@apache.org>.
nipunbatra8 commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2308323694
##########
lucene/sandbox/src/java/org/apache/lucene/sandbox/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+import org.apache.lucene.index.MergeScheduler;
+
+/**
+ * A {@link MergeScheduler} that caps total IO write bandwidth across all running merges to a
+ * specified max MB/sec bandwidth.
+ *
+ * @lucene.experimental
+ */
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Global bandwidth cap in MB/s. Mutable so that updates are applied live. */
+ private double bandwidthMbPerSec;
+
+ /** Create a scheduler with a required global bandwidth cap in MB/s. */
+ public BandwidthCappedMergeScheduler(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
Review Comment:
Good point
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "shubhamsrkdev (via GitHub)" <gi...@apache.org>.
shubhamsrkdev commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2307033186
##########
lucene/sandbox/src/test/org/apache/lucene/sandbox/index/TestBandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import com.carrotsearch.randomizedtesting.generators.RandomStrings;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.document.TextField;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
+import org.apache.lucene.index.LogDocMergePolicy;
+import org.apache.lucene.index.MergePolicy;
+import org.apache.lucene.index.MergeScheduler;
+import org.apache.lucene.store.Directory;
+import org.apache.lucene.tests.util.LuceneTestCase;
+import org.apache.lucene.util.InfoStream;
+
+/** Tests for {@link BandwidthCappedMergeScheduler}. */
+public class TestBandwidthCappedMergeScheduler extends LuceneTestCase {
+
+ public void testBasicFunctionality() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler(100.0);
+
+ // Test initial value
+ assertEquals(100.0, scheduler.getMaxMbPerSec(), 0.001);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testBandwidthConfiguration() throws Exception {
+ Directory dir = newDirectory();
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler(50.0);
+
+ // Test setting valid bandwidth rates
+ scheduler.setMaxMbPerSec(200.0);
+ assertEquals(200.0, scheduler.getMaxMbPerSec(), 0.001);
+
+ scheduler.setMaxMbPerSec(10.0);
+ assertEquals(10.0, scheduler.getMaxMbPerSec(), 0.001);
+
+ scheduler.close();
+ dir.close();
+ }
+
+ public void testInvalidBandwidthConfiguration() throws Exception {
+ // Test invalid constructor parameter
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ new BandwidthCappedMergeScheduler(0.0);
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ new BandwidthCappedMergeScheduler(-1.0);
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ new BandwidthCappedMergeScheduler(Double.NaN);
+ });
+
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler(100.0);
+
+ // Test invalid setMaxMbPerSec values
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setMaxMbPerSec(0.0);
+ });
+
+ expectThrows(
+ IllegalArgumentException.class,
+ () -> {
+ scheduler.setMaxMbPerSec(-1.0);
+ });
+
+ scheduler.close();
+ }
+
+ public void testToString() throws Exception {
+ BandwidthCappedMergeScheduler scheduler = new BandwidthCappedMergeScheduler(50.0);
+ String str = scheduler.toString();
+
+ assertTrue(str.contains("BandwidthCappedMergeScheduler"));
+ assertTrue(str.contains("bandwidthMbPerSec"));
+ assertTrue(str.contains("MB/s"));
+ assertTrue(str.contains("50.0"));
+
+ scheduler.close();
+ }
+
+ public void testWithIndexWriter() throws IOException {
Review Comment:
I am not sure about the usability of this test; we are checking in assertion `scheduler.getMaxMbPerSec()` which returns `bandwidthMbPerSec` which was set in the constructor. The rest of the logic (document creation, write, commit, merges) don't influence the `bandwidthMbPerSec` value in any way. Wouldn't this make this test very similar to the first test (setting and getting `bandwidthMbPerSec`)?
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "nipunbatra8 (via GitHub)" <gi...@apache.org>.
nipunbatra8 commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2252124214
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
Review Comment:
Good point, I think something closer to START_BANDWIDTH_RATE would be better.
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "shubhamsrkdev (via GitHub)" <gi...@apache.org>.
shubhamsrkdev commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2244963804
##########
lucene/core/src/java/org/apache/lucene/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+
+/**
+ * A {@link MergeScheduler} that extends {@link ConcurrentMergeScheduler} with bandwidth tracking
+ * and limiting capabilities. This scheduler maintains a bandwidth rate bucket that is divided among
+ * active merges. When bandwidth is limited, merges are throttled to preserve system resources.
+ *
+ * <p>Key features: Global bandwidth rate bucket with configurable capacity, dynamic per-merge
+ * throttling.
+ *
+ * @lucene.experimental
+ */
+// nocommit
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Floor for IO write rate limit (we will never go any lower than this) */
+ private static final double MIN_MERGE_MB_PER_SEC = 5.0;
+
+ /** Ceiling for IO write rate limit (we will never go any higher than this) */
+ private static final double MAX_MERGE_MB_PER_SEC = 10240.0;
+
+ /** Initial value for IO write rate limit */
+ private static final double START_MB_PER_SEC = 1000.0;
+
+ /** Global bandwidth rate bucket in MB/s */
+ private double bandwidthRateBucket = START_MB_PER_SEC;
+
+ /** Default constructor with 1000 MB/s bandwidth rate bucket */
+ public BandwidthCappedMergeScheduler() {
+ super();
+ }
+
+ /** Set the global bandwidth rate bucket in MB/s (default 1000 MB/s) */
+ public void setBandwidthRateBucket(double mbPerSec) {
+ if (mbPerSec < MIN_MERGE_MB_PER_SEC || mbPerSec > MAX_MERGE_MB_PER_SEC) {
Review Comment:
nit: Bracket indentation.
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14964:
URL: https://github.com/apache/lucene/pull/14964#issuecomment-3300766072
This PR has not had activity in the past 2 weeks, labeling it as stale. If the PR is waiting for review, notify the dev@lucene.apache.org list. Thank you for your contribution!
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org
Re: [PR] [Draft] Add BandwidthCappedMergeScheduler for enforcing a global merge bandwidth cap [lucene]
Posted by "shubhamsrkdev (via GitHub)" <gi...@apache.org>.
shubhamsrkdev commented on code in PR #14964:
URL: https://github.com/apache/lucene/pull/14964#discussion_r2306947911
##########
lucene/sandbox/src/java/org/apache/lucene/sandbox/index/BandwidthCappedMergeScheduler.java:
##########
@@ -0,0 +1,185 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.lucene.sandbox.index;
+
+import java.io.IOException;
+import org.apache.lucene.index.ConcurrentMergeScheduler;
+import org.apache.lucene.index.MergePolicy.OneMerge;
+import org.apache.lucene.index.MergeScheduler;
+
+/**
+ * A {@link MergeScheduler} that caps total IO write bandwidth across all running merges to a
+ * specified max MB/sec bandwidth.
+ *
+ * @lucene.experimental
+ */
+public class BandwidthCappedMergeScheduler extends ConcurrentMergeScheduler {
+
+ /** Global bandwidth cap in MB/s. Mutable so that updates are applied live. */
+ private double bandwidthMbPerSec;
+
+ /** Create a scheduler with a required global bandwidth cap in MB/s. */
+ public BandwidthCappedMergeScheduler(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
+ || Double.isNaN(bandwidthMbPerSec)
+ || Double.isInfinite(bandwidthMbPerSec)) {
+ throw new IllegalArgumentException("bandwidthMbPerSec must be a finite positive value");
+ }
+ this.bandwidthMbPerSec = bandwidthMbPerSec;
+ }
+
+ /**
+ * Auto IO throttling is managed by this scheduler's bandwidth cap. Enabling parent CMS IO
+ * throttling is ignored.
+ */
+ @Override
+ public synchronized void enableAutoIOThrottle() {
+ if (verbose()) {
+ message("Ignoring enableAutoIOThrottle; using bandwidth cap instead");
+ }
+ // Intentionally no-op
+ }
+
+ /** Ensure auto IO throttling remains disabled. */
+ @Override
+ public synchronized void disableAutoIOThrottle() {
+ // Make sure parent state is disabled if it was somehow enabled earlier
+ super.disableAutoIOThrottle();
+ }
+
+ /** Always returns false since CMS auto IO throttling is disabled for this scheduler. */
+ @Override
+ public synchronized boolean getAutoIOThrottle() {
+ return false;
+ }
+
+ /** Get the global bandwidth cap in MB/s */
+ public double getMaxMbPerSec() {
+ return bandwidthMbPerSec;
+ }
+
+ /**
+ * Set the global bandwidth cap in MB/s.
+ *
+ * <p>This setting is live: merges that are already running will be adjusted to the new per-merge
+ * rate derived from this cap, without requiring a restart.
+ *
+ * @param bandwidthMbPerSec the new global bandwidth cap in MB/s; must be finite and > 0
+ */
+ public synchronized void setMaxMbPerSec(double bandwidthMbPerSec) {
+ if (!(bandwidthMbPerSec > 0.0)
Review Comment:
I see we are checking the same thing in constructor also, maybe worth creating a separate private method for this?
--
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: issues-unsubscribe@lucene.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@lucene.apache.org
For additional commands, e-mail: issues-help@lucene.apache.org