diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java index 649b3bd3df..ce09905999 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/SimplePool.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2021 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -48,7 +48,7 @@ public class SimplePool implements Pool { protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR final - private final Semaphore permits = new Semaphore(0); + private final PoolSemaphore permits = new PoolSemaphore(0); private final AtomicInteger poolSize = new AtomicInteger(); @@ -56,11 +56,11 @@ public class SimplePool implements Pool { private long waitTimeout = Long.MAX_VALUE; - private final BlockingQueue available = new LinkedBlockingQueue(); + private final BlockingQueue available = new LinkedBlockingQueue<>(); - private final Set allocated = Collections.synchronizedSet(new HashSet()); + private final Set allocated = Collections.synchronizedSet(new HashSet<>()); - private final Set inUse = Collections.synchronizedSet(new HashSet()); + private final Set inUse = Collections.synchronizedSet(new HashSet<>()); private final PoolItemCallback callback; @@ -105,21 +105,27 @@ public class SimplePool implements Pool { this.permits.release(delta); } else { - while (delta < 0) { - if (!this.permits.tryAcquire()) { - break; - } + this.permits.reducePermits(-delta); + + int inUseSize = this.inUse.size(); + int newPoolSize = Math.max(poolSize, inUseSize); + this.poolSize.set(newPoolSize); + + for (int i = this.available.size(); i > newPoolSize - inUseSize; i--) { T item = this.available.poll(); if (item != null) { doRemoveItem(item); } - this.poolSize.decrementAndGet(); - delta++; + else { + break; + } + } + + int inUseDelta = poolSize - inUseSize; + if (inUseDelta < 0 && this.logger.isDebugEnabled()) { + this.logger.debug(String.format("Pool is overcommitted by %d; items will be removed when returned", + -inUseDelta)); } - } - if (delta < 0 && this.logger.isDebugEnabled()) { - this.logger.debug(String.format("Pool is overcommitted by %d; items will be removed when returned", - -delta)); } } @@ -266,6 +272,19 @@ public class SimplePool implements Pool { removeAllIdleItems(); } + private static class PoolSemaphore extends Semaphore { + + PoolSemaphore(int permits) { + super(permits); + } + + @Override + public void reducePermits(int reduction) { // NOSONAR increases visibility + super.reducePermits(reduction); + } + + } + /** * User of the pool provide an implementation of this interface; called during * various pool operations. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java b/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java index 883390f568..332fdbc368 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/util/SimplePoolTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2022 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -20,7 +20,6 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; import static org.assertj.core.api.Assertions.assertThatIllegalStateException; -import static org.assertj.core.api.Assertions.fail; import java.util.ArrayList; import java.util.HashSet; @@ -36,6 +35,8 @@ import org.springframework.integration.test.util.TestUtils; /** * @author Gary Russell * @author Sergey Bogatyrev + * @author Artem Bilan + * * @since 2.2 * */ @@ -43,7 +44,7 @@ public class SimplePoolTests { @Test public void testReuseAndStale() { - final Set strings = new HashSet(); + final Set strings = new HashSet<>(); final AtomicBoolean stale = new AtomicBoolean(); SimplePool pool = stringPool(2, strings, stale); String s1 = pool.getItem(); @@ -62,7 +63,7 @@ public class SimplePoolTests { @Test public void testOverCommitAndResize() { - final Set strings = new HashSet(); + final Set strings = new HashSet<>(); final AtomicBoolean stale = new AtomicBoolean(); SimplePool pool = stringPool(2, strings, stale); String s1 = pool.getItem(); @@ -83,13 +84,9 @@ public class SimplePoolTests { assertThat(pool.getIdleCount()).isEqualTo(0); assertThat(pool.getActiveCount()).isEqualTo(2); assertThat(pool.getAllocatedCount()).isEqualTo(2); - try { - pool.getItem(); - fail("Expected exception"); - } - catch (PoolItemNotAvailableException e) { - } + assertThatExceptionOfType(PoolItemNotAvailableException.class) + .isThrownBy(pool::getItem); // resize up pool.setPoolSize(4); @@ -131,7 +128,7 @@ public class SimplePoolTests { @Test public void testForeignObject() { - final Set strings = new HashSet(); + final Set strings = new HashSet<>(); final AtomicBoolean stale = new AtomicBoolean(); SimplePool pool = stringPool(2, strings, stale); pool.getItem(); @@ -140,7 +137,7 @@ public class SimplePoolTests { @Test public void testDoubleReturn() { - final Set strings = new HashSet(); + final Set strings = new HashSet<>(); final AtomicBoolean stale = new AtomicBoolean(); SimplePool pool = stringPool(2, strings, stale); Semaphore permits = TestUtils.getPropertyValue(pool, "permits", Semaphore.class); @@ -168,7 +165,27 @@ public class SimplePoolTests { assertThat(allocatedItems).hasSize(5); // no more items can be allocated (indirect check of permits) - assertThatExceptionOfType(PoolItemNotAvailableException.class).isThrownBy(() -> pool.getItem()); + assertThatExceptionOfType(PoolItemNotAvailableException.class) + .isThrownBy(pool::getItem); + } + + @Test + public void testMaxValueSizeUpdateIfNotAllocated() { + SimplePool pool = stringPool(0, new HashSet<>(), new AtomicBoolean()); + pool.setWaitTimeout(0); + pool.setPoolSize(5); + assertThat(pool.getPoolSize()).isEqualTo(5); + + // allocating all available items to check permits + Set allocatedItems = new HashSet<>(); + for (int i = 0; i < 5; i++) { + allocatedItems.add(pool.getItem()); + } + assertThat(allocatedItems).hasSize(5); + + // no more items can be allocated (indirect check of permits) + assertThatExceptionOfType(PoolItemNotAvailableException.class) + .isThrownBy(pool::getItem); } @Test @@ -207,7 +224,54 @@ public class SimplePoolTests { assertThat(pool.getActiveCount()).isEqualTo(5); // no more items can be allocated (indirect check of permits) - assertThatExceptionOfType(PoolItemNotAvailableException.class).isThrownBy(() -> pool.getItem()); + assertThatExceptionOfType(PoolItemNotAvailableException.class) + .isThrownBy(pool::getItem); + } + + @Test + public void testMaxValueSizeUpdateIfPartiallyAllocated() { + SimplePool pool = stringPool(0, new HashSet<>(), new AtomicBoolean()); + pool.setWaitTimeout(0); + + List allocated = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + allocated.add(pool.getItem()); + } + + // release only 2 items + for (int i = 0; i < 2; i++) { + pool.releaseItem(allocated.get(i)); + } + + // release only 2 items + for (int i = 0; i < 2; i++) { + pool.releaseItem(allocated.get(i)); + } + + // trying to reduce pool size + pool.setPoolSize(5); + + // at this moment the actual pool size can be reduced only partially, because + // only 2 items have been released, so 8 items are in use + assertThat(pool.getPoolSize()).isEqualTo(8); + assertThat(pool.getAllocatedCount()).isEqualTo(8); + assertThat(pool.getIdleCount()).isEqualTo(0); + assertThat(pool.getActiveCount()).isEqualTo(8); + + // releasing 3 items + for (int i = 2; i < 5; i++) { + pool.releaseItem(allocated.get(i)); + } + + // now pool size should be reduced + assertThat(pool.getPoolSize()).isEqualTo(5); + assertThat(pool.getAllocatedCount()).isEqualTo(5); + assertThat(pool.getIdleCount()).isEqualTo(0); + assertThat(pool.getActiveCount()).isEqualTo(5); + + // no more items can be allocated (indirect check of permits) + assertThatExceptionOfType(PoolItemNotAvailableException.class) + .isThrownBy(pool::getItem); } @Test @@ -240,7 +304,8 @@ public class SimplePoolTests { assertThat(pool.getActiveCount()).isEqualTo(5); // no more items can be allocated (indirect check of permits) - assertThatExceptionOfType(PoolItemNotAvailableException.class).isThrownBy(() -> pool.getItem()); + assertThatExceptionOfType(PoolItemNotAvailableException.class) + .isThrownBy(pool::getItem); // releasing remaining items for (int i = 5; i < 10; i++) { @@ -266,10 +331,9 @@ public class SimplePoolTests { assertThatIllegalStateException().isThrownBy(pool::getItem); } - private SimplePool stringPool(int size, final Set strings, - final AtomicBoolean stale) { + private SimplePool stringPool(int size, Set strings, AtomicBoolean stale) { + return new SimplePool(size, new SimplePool.PoolItemCallback() { - SimplePool pool = new SimplePool(size, new SimplePool.PoolItemCallback() { private int i; @Override @@ -293,7 +357,6 @@ public class SimplePoolTests { } }); - return pool; } }