This is an automated email from the ASF dual-hosted git repository. rmaucher pushed a commit to branch 10.1.x in repository https://gitbox.apache.org/repos/asf/tomcat.git
commit 3a05278fa06646c996669a676c2d1166273beb6d Author: opencode <[email protected]> AuthorDate: Sun Oct 11 10:13:38 2026 +0200 Make the buffer pool size accounting atomic The pool size limit was enforced with a check against the current total followed by a separate increment, so concurrent returns could all pass the check against the same stale total and admit more bytes between them than the configured maximum. Clearing the pool was not atomic either. A return that interleaved between the queue clear and the size reset left the accounting permanently skewed, a buffer kept in the queue without being counted, or counted without being kept, biasing every later admission decision over or under the limit. The limit is now enforced with a compare-and-set loop, making the size never exceed the maximum under concurrent returns, and the clear now removes each buffer from the queue and its capacity from the total as one paired operation, so the size again matches the buffers that the queue holds once the pool is quiescent. This also removes a related skew where a subtraction landing after the reset could drive the size negative. Add tests that race returns against the limit and against a concurrent clear, both of which fail reliably on the previous code. --- java/org/apache/catalina/tribes/io/BufferPool.java | 27 +++- .../apache/catalina/tribes/io/TestBufferPool.java | 153 +++++++++++++++++++++ 2 files changed, 174 insertions(+), 6 deletions(-) diff --git a/java/org/apache/catalina/tribes/io/BufferPool.java b/java/org/apache/catalina/tribes/io/BufferPool.java index fae2798cb1..7b4656c5b9 100644 --- a/java/org/apache/catalina/tribes/io/BufferPool.java +++ b/java/org/apache/catalina/tribes/io/BufferPool.java @@ -90,16 +90,31 @@ public class BufferPool { * @param buffer the buffer to return */ public void returnBuffer(XByteBuffer buffer) { - if ((size.get() + buffer.getCapacity()) <= maxSize) { - size.addAndGet(buffer.getCapacity()); - queue.offer(buffer); - } + // the pool size limit is enforced with a compare-and-set loop so that + // concurrent returns cannot all pass the limit check against the same + // stale total and admit more than maxSize bytes between them + int capacity = buffer.getCapacity(); + int current; + do { + current = size.get(); + if ((long) current + capacity > maxSize) { + return; + } + } while (!size.compareAndSet(current, current + capacity)); + queue.offer(buffer); } /** Clears all buffers from the pool. */ public void clear() { - queue.clear(); - size.set(0); + // drain the queue buffer by buffer, removing each capacity from the + // size total as the buffer leaves the queue; setting the size to zero + // separately would race with concurrent returns and skew the + // accounting permanently + XByteBuffer buffer = queue.poll(); + while (buffer != null) { + size.addAndGet(-buffer.getCapacity()); + buffer = queue.poll(); + } } /** Maximum pool size in bytes. */ diff --git a/test/org/apache/catalina/tribes/io/TestBufferPool.java b/test/org/apache/catalina/tribes/io/TestBufferPool.java new file mode 100644 index 0000000000..2ed916c71d --- /dev/null +++ b/test/org/apache/catalina/tribes/io/TestBufferPool.java @@ -0,0 +1,153 @@ +/* + * 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.catalina.tribes.io; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Assert; +import org.junit.Test; + +public class TestBufferPool { + + private static final int MAX = 512 * 1024; + + /* + * Concurrent returns must not all pass the limit check against the same stale total and admit + * more than maxSize bytes between them. The pool is parked just below the limit and repeatedly + * offered to the racing threads, so the limit check is exercised over and over on the edge. + */ + @Test + public void testConcurrentReturnsRespectMaxSize() throws Exception { + BufferPool pool = BufferPool.getBufferPool(); + int original = pool.getMaxSize(); + try { + pool.clear(); + pool.setMaxSize(MAX); + final AtomicBoolean go = new AtomicBoolean(); + final AtomicBoolean stop = new AtomicBoolean(); + List<Thread> threads = new ArrayList<>(); + try { + for (int t = 0; t < 4; t++) { + final XByteBuffer[] ring = new XByteBuffer[32]; + for (int i = 0; i < ring.length; i++) { + ring[i] = new XByteBuffer(64 * 1024, false); + } + Thread thread = new Thread(() -> { + int index = 0; + while (!stop.get()) { + while (!go.get() && !stop.get()) { + Thread.onSpinWait(); + } + while (go.get() && !stop.get()) { + pool.returnBuffer(ring[index++ & 31]); + } + } + }); + thread.setDaemon(true); + threads.add(thread); + thread.start(); + } + int overshoot = 0; + int trials = 750; + for (int trial = 0; trial < trials; trial++) { + drain(pool); + for (int i = 0; i < MAX / (64 * 1024) - 1; i++) { + pool.returnBuffer(new XByteBuffer(64 * 1024, false)); + } + go.set(true); + Thread.sleep(0, 200000); + go.set(false); + if (pool.size.get() > MAX) { + overshoot++; + } + } + Assert.assertEquals("pool size exceeded maxSize in " + overshoot + " of the concurrent trials", 0, + overshoot); + } finally { + stop.set(true); + for (Thread thread : threads) { + thread.join(); + } + } + } finally { + pool.setMaxSize(original); + pool.clear(); + } + } + + private static void drain(BufferPool pool) { + while (!pool.queue.isEmpty()) { + pool.getBuffer(1, false); + } + } + + /* + * Clearing the pool while buffers are returned must not skew the size accounting. Once all + * activity has stopped, the size must again match the buffers that the queue actually holds. + */ + @Test + public void testClearKeepsAccountingConsistent() throws Exception { + BufferPool pool = BufferPool.getBufferPool(); + int original = pool.getMaxSize(); + try { + pool.clear(); + pool.setMaxSize(MAX); + AtomicBoolean stop = new AtomicBoolean(); + Thread returns = new Thread(() -> { + while (!stop.get()) { + pool.returnBuffer(new XByteBuffer(1024, false)); + } + }); + returns.setDaemon(true); + Thread[] clears = new Thread[2]; + try { + returns.start(); + for (int t = 0; t < clears.length; t++) { + clears[t] = new Thread(() -> { + while (!stop.get()) { + pool.clear(); + } + }); + clears[t].setDaemon(true); + clears[t].start(); + } + Thread.sleep(250); + } finally { + stop.set(true); + returns.join(); + for (Thread thread : clears) { + thread.join(); + } + } + // Once all activity has stopped, the size must again match the + // buffers that the queue actually holds + long queued = 0; + for (XByteBuffer buffer : pool.queue) { + queued += buffer.getCapacity(); + } + Assert.assertEquals("pool size must match the capacity of the queued buffers", queued, pool.size.get()); + pool.clear(); + Assert.assertEquals(0, pool.size.get()); + Assert.assertTrue(pool.queue.isEmpty()); + } finally { + pool.setMaxSize(original); + pool.clear(); + } + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
