diff --git a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java index 53e4a963..b34ed942 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/DefaultTaskito.java @@ -41,6 +41,7 @@ import org.byteveda.taskito.predicates.EnqueueGate; import org.byteveda.taskito.predicates.Predicate; import org.byteveda.taskito.predicates.PredicateContext; +import org.byteveda.taskito.resources.PoolConfig; import org.byteveda.taskito.resources.ResourceContext; import org.byteveda.taskito.resources.ResourceDefinition; import org.byteveda.taskito.resources.ResourceRuntime; @@ -116,6 +117,16 @@ public Taskito resource( return this; } + @Override + public Taskito resource( + String name, PoolConfig pool, Function factory, Consumer dispose) { + resources.register( + name, + new ResourceDefinition( + factory::apply, ResourceScope.POOLED, value -> dispose.accept(cast(value)), pool)); + return this; + } + @Override public Map resourceMetrics() { return resources.metrics(); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java index 95dc280f..b54583c2 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/Taskito.java @@ -32,6 +32,7 @@ import org.byteveda.taskito.model.WorkerInfo; import org.byteveda.taskito.predicates.EnqueueGate; import org.byteveda.taskito.predicates.Predicate; +import org.byteveda.taskito.resources.PoolConfig; import org.byteveda.taskito.resources.ResourceContext; import org.byteveda.taskito.resources.ResourceScope; import org.byteveda.taskito.resources.ResourceStat; @@ -78,6 +79,15 @@ static Builder builder() { /** Register a resource with a scope and a disposer run when the scope ends. */ Taskito resource(String name, ResourceScope scope, Function factory, Consumer dispose); + /** + * Register a {@link ResourceScope#POOLED} resource: a bounded pool of + * instances shared across tasks. Each task checks one instance out for its + * duration and returns it at task end; {@code pool} bounds capacity and + * {@code dispose} runs when the pool retires an instance (worker shutdown or + * {@link PoolConfig#maxLifetime()} expiry). + */ + Taskito resource(String name, PoolConfig pool, Function factory, Consumer dispose); + /** Per-resource counters (created / disposed / active). */ Map resourceMetrics(); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/PoolConfig.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/PoolConfig.java new file mode 100644 index 00000000..341251e0 --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/PoolConfig.java @@ -0,0 +1,60 @@ +package org.byteveda.taskito.resources; + +import java.time.Duration; + +/** + * Sizing and timing for a {@link ResourceScope#POOLED} resource (part of the + * cross-SDK contract for pooled resources). + * + * @param poolSize the maximum number of concurrently checked-out instances (must be > 0) + * @param poolMin how many instances to build eagerly at worker start (defaults to 0, at most + * {@code poolSize}) + * @param acquireTimeout how long a checkout may wait for capacity (defaults to 10s) + * @param maxLifetime an idle instance older than this is disposed instead of reused + * ({@code null} keeps instances forever) + */ +public record PoolConfig(int poolSize, int poolMin, Duration acquireTimeout, Duration maxLifetime) { + + private static final Duration DEFAULT_ACQUIRE_TIMEOUT = Duration.ofSeconds(10); + + public PoolConfig { + if (poolSize <= 0) { + throw new IllegalArgumentException("poolSize must be > 0"); + } + if (poolMin < 0) { + throw new IllegalArgumentException("poolMin must be >= 0"); + } + if (poolMin > poolSize) { + throw new IllegalArgumentException("poolMin must be <= poolSize"); + } + if (acquireTimeout == null) { + acquireTimeout = DEFAULT_ACQUIRE_TIMEOUT; + } + if (acquireTimeout.isNegative() || acquireTimeout.isZero()) { + throw new IllegalArgumentException("acquireTimeout must be positive"); + } + if (maxLifetime != null && (maxLifetime.isNegative() || maxLifetime.isZero())) { + throw new IllegalArgumentException("maxLifetime must be positive"); + } + } + + /** A pool of {@code poolSize} with no prewarm, a 10s acquire timeout, and unlimited lifetime. */ + public static PoolConfig of(int poolSize) { + return new PoolConfig(poolSize, 0, null, null); + } + + /** This configuration with {@code poolMin} instances built eagerly at worker start. */ + public PoolConfig withPoolMin(int poolMin) { + return new PoolConfig(poolSize, poolMin, acquireTimeout, maxLifetime); + } + + /** This configuration with a different checkout wait limit. */ + public PoolConfig withAcquireTimeout(Duration acquireTimeout) { + return new PoolConfig(poolSize, poolMin, acquireTimeout, maxLifetime); + } + + /** This configuration with a maximum instance lifetime ({@code null} for unlimited). */ + public PoolConfig withMaxLifetime(Duration maxLifetime) { + return new PoolConfig(poolSize, poolMin, acquireTimeout, maxLifetime); + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceContext.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceContext.java index f18119e8..c1bd8148 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceContext.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceContext.java @@ -2,8 +2,9 @@ /** * Handed to a resource factory so it can depend on other resources. A factory - * may only depend on same-or-longer-lived resources: a {@code WORKER} factory - * may {@link #use} only worker resources, a {@code THREAD} factory may use + * may only depend on same-or-longer-lived resources: {@code WORKER} and + * {@code POOLED} factories may {@link #use} only worker resources (a pooled + * instance outlives the task that built it), a {@code THREAD} factory may use * worker or thread resources, and {@code TASK}/{@code REQUEST} factories may * use any scope. */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceDefinition.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceDefinition.java index b32f5dbb..f439cd3d 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceDefinition.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceDefinition.java @@ -9,9 +9,11 @@ * @param factory builds the resource, possibly using others via the context * @param scope the resource's lifetime (defaults to {@link ResourceScope#WORKER}) * @param dispose optional cleanup run when the scope ends ({@code null} for none) + * @param pool bounded-pool sizing, required for {@link ResourceScope#POOLED} and + * {@code null} for every other scope */ public record ResourceDefinition( - Function factory, ResourceScope scope, Consumer dispose) { + Function factory, ResourceScope scope, Consumer dispose, PoolConfig pool) { public ResourceDefinition { if (factory == null) { @@ -20,6 +22,18 @@ public record ResourceDefinition( if (scope == null) { scope = ResourceScope.WORKER; } + if (scope == ResourceScope.POOLED && pool == null) { + throw new IllegalArgumentException("a pooled resource requires a PoolConfig"); + } + if (scope != ResourceScope.POOLED && pool != null) { + throw new IllegalArgumentException("a PoolConfig is only valid for a pooled resource"); + } + } + + /** A definition for any non-pooled scope. */ + public ResourceDefinition( + Function factory, ResourceScope scope, Consumer dispose) { + this(factory, scope, dispose, null); } /** A worker-scoped resource with no disposer. */ diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourcePool.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourcePool.java new file mode 100644 index 00000000..4917d33d --- /dev/null +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourcePool.java @@ -0,0 +1,231 @@ +package org.byteveda.taskito.resources; + +import java.lang.System.Logger; +import java.lang.System.Logger.Level; +import java.time.Duration; +import java.util.ArrayDeque; +import java.util.Deque; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Consumer; +import java.util.function.Supplier; +import org.byteveda.taskito.errors.ResourceException; + +/** + * A bounded checkout/return pool backing {@link ResourceScope#POOLED} resources + * (part of the cross-SDK contract for pooled resources). A semaphore caps + * concurrent checkouts at {@link PoolConfig#poolSize()}; released instances park + * on an idle deque and are reused unless {@link PoolConfig#maxLifetime()} has + * elapsed, in which case they are disposed and a fresh instance is built. The + * factory always runs on the acquiring thread (so per-thread bookkeeping such as + * dependency-cycle detection keeps working). Thread-safe. + */ +public final class ResourcePool { + private static final Logger LOG = System.getLogger(ResourcePool.class.getName()); + /** Sentinel distinguishing "no idle instance" from a factory that returned {@code null}. */ + private static final Object NO_IDLE = new Object(); + + private final String name; + private final PoolConfig config; + private final Supplier factory; + private final Consumer dispose; + + private final Semaphore capacity; + private final Deque idle = new ArrayDeque<>(); + private final ReentrantLock lock = new ReentrantLock(); + private int active; // guarded by lock + private long totalAcquisitions; // guarded by lock + private long totalTimeouts; // guarded by lock + private boolean closed; // guarded by lock + + /** An instance parked in the pool, stamped so {@code maxLifetime} can expire it. */ + private record IdleInstance(Object instance, long createdAtNanos) { + static IdleInstance of(Object instance) { + return new IdleInstance(instance, System.nanoTime()); + } + } + + /** + * A point-in-time snapshot of the pool. + * + * @param size the configured capacity + * @param active instances currently checked out + * @param idle instances parked and ready for reuse + * @param totalAcquisitions successful checkouts so far + * @param totalTimeouts checkouts that gave up waiting for capacity + */ + public record Stats(int size, int active, int idle, long totalAcquisitions, long totalTimeouts) {} + + /** + * @param name the resource name (used in error messages and logs) + * @param config pool sizing and timing + * @param factory builds one instance; invoked on the acquiring thread + * @param dispose tears one instance down, or {@code null} for none + */ + public ResourcePool(String name, PoolConfig config, Supplier factory, Consumer dispose) { + this.name = name; + this.config = config; + this.factory = factory; + this.dispose = dispose; + this.capacity = new Semaphore(config.poolSize()); + } + + /** Build {@link PoolConfig#poolMin()} instances upfront, best-effort: log and stop on the first failure. */ + public void prewarm() { + for (int i = 0; i < config.poolMin(); i++) { + Object instance; + try { + instance = factory.get(); + } catch (RuntimeException e) { + LOG.log(Level.WARNING, "failed to prewarm resource '" + name + "'", e); + return; + } + lock.lock(); + try { + if (closed) { + // A shutdown raced the prewarm — don't park instances on a + // drained pool; dispose the fresh build and stop. + disposeQuietly(instance); + return; + } + idle.addLast(IdleInstance.of(instance)); + } finally { + lock.unlock(); + } + } + } + + /** + * Check an instance out, blocking up to {@link PoolConfig#acquireTimeout()} + * for capacity. Reuses an idle instance when one is still within its + * lifetime; otherwise builds a fresh one on the calling thread. + * + * @throws ResourceException when the pool stays exhausted past the timeout + */ + public Object acquire() { + awaitCapacity(); + Object reused = pollIdle(); + if (reused != NO_IDLE) { + return reused; + } + // No idle instance — build one, holding the permit throughout. The active + // count moves only after the factory succeeds, so a failing factory can + // neither leak capacity nor underflow the counter. + Object instance; + try { + instance = factory.get(); + } catch (RuntimeException | Error e) { + capacity.release(); + throw e; + } + lock.lock(); + try { + recordAcquired(); + } finally { + lock.unlock(); + } + return instance; + } + + /** Return a checked-out instance: park it for reuse, or dispose it if the pool has shut down. */ + public void release(Object instance) { + lock.lock(); + try { + active--; + if (closed) { + disposeQuietly(instance); + } else { + idle.addLast(IdleInstance.of(instance)); + } + } finally { + lock.unlock(); + } + capacity.release(); + } + + /** Dispose every idle instance and mark the pool shut down (later releases dispose immediately). */ + public void shutdown() { + lock.lock(); + try { + closed = true; + while (!idle.isEmpty()) { + disposeQuietly(idle.pollFirst().instance()); + } + } finally { + lock.unlock(); + } + } + + /** A snapshot of the pool's counters. */ + public Stats stats() { + lock.lock(); + try { + return new Stats(config.poolSize(), active, idle.size(), totalAcquisitions, totalTimeouts); + } finally { + lock.unlock(); + } + } + + /** Take a capacity permit or fail with a timeout error. */ + private void awaitCapacity() { + boolean acquired; + try { + acquired = capacity.tryAcquire(config.acquireTimeout().toNanos(), TimeUnit.NANOSECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new ResourceException("interrupted while acquiring resource '" + name + "' from its pool", e); + } + if (!acquired) { + lock.lock(); + try { + totalTimeouts++; + } finally { + lock.unlock(); + } + throw new ResourceException("resource '" + name + "' pool timed out after " + + config.acquireTimeout().toMillis() + "ms"); + } + } + + /** Pop a live idle instance, disposing any that outlived {@code maxLifetime}; {@link #NO_IDLE} if none. */ + private Object pollIdle() { + lock.lock(); + try { + while (!idle.isEmpty()) { + IdleInstance entry = idle.pollFirst(); + if (!expired(entry)) { + recordAcquired(); + return entry.instance(); + } + disposeQuietly(entry.instance()); + } + return NO_IDLE; + } finally { + lock.unlock(); + } + } + + private boolean expired(IdleInstance entry) { + Duration lifetime = config.maxLifetime(); + return lifetime != null && System.nanoTime() - entry.createdAtNanos() >= lifetime.toNanos(); + } + + /** Update counters for a successful checkout. Caller holds {@code lock}. */ + private void recordAcquired() { + totalAcquisitions++; + active++; + } + + /** Dispose one instance; a failing disposer is logged, never propagated. */ + private void disposeQuietly(Object instance) { + if (dispose == null) { + return; + } + try { + dispose.accept(instance); + } catch (RuntimeException e) { + LOG.log(Level.WARNING, "disposing pooled resource '" + name + "' failed", e); + } + } +} diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java index 030cc188..89938148 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceRuntime.java @@ -41,6 +41,12 @@ public final class ResourceRuntime { * the concurrent maps exist for cross-thread visibility at worker teardown. */ private final ConcurrentMap> threadCache = new ConcurrentHashMap<>(); + /** + * Per-name pools for {@code POOLED}-scoped resources. An instance field like + * {@code workerCache}, so each worker runtime from {@link #forWorker()} owns + * its own pools — capacity is per worker, never shared across workers. + */ + private final ConcurrentMap pools = new ConcurrentHashMap<>(); private final Deque workerTeardown = new ArrayDeque<>(); /** Names being resolved on the current thread, so a dependency cycle fails fast instead of recursing. */ @@ -74,6 +80,23 @@ public T use(String name) { } }; + /** + * Context handed to a pooled factory: it may only use worker resources — + * pooled instances outlive tasks, so a task/request/thread-scoped dependency + * would dangle after its own scope ends. + */ + private final ResourceContext pooledContext = new ResourceContext() { + @Override + public ResourceScope scope() { + return ResourceScope.POOLED; + } + + @Override + public T use(String name) { + return cast(resolvePooledDependency(name)); + } + }; + /** A client-level runtime: holds definitions + counters, hands each worker a child via {@link #forWorker()}. */ public ResourceRuntime() { this.definitions = new ConcurrentHashMap<>(); @@ -113,9 +136,19 @@ public TaskScope createTaskScope() { return new TaskScope(this); } - /** Lease the worker resources (paired with {@link #teardownWorker}). */ - public synchronized void acquireWorker() { - leases++; + /** Lease the worker resources (paired with {@link #teardownWorker}); the first lease prewarms pools. */ + public void acquireWorker() { + boolean firstLease; + synchronized (this) { + firstLease = leases == 0; + leases++; + } + // Outside the monitor: prewarm runs user factories, which may be slow — + // holding the lock would stall every concurrent lease/teardown. A racing + // shutdown is safe: the pool disposes prewarmed instances once closed. + if (firstLease) { + prewarmPools(); + } } /** Release a worker lease; when the last one drops, dispose worker resources LIFO. */ @@ -175,7 +208,8 @@ Object resolveWorker(String name) { /** * Resolve for a task, dispatching by the resource's scope: worker hits the * shared cache, thread hits the current thread's cache, request builds fresh - * on every use, task builds once per invocation. + * on every use, pooled checks an instance out for the invocation, task builds + * once per invocation. */ Object resolveForTask(TaskScope scope, String name) { ResourceDefinition definition = definition(name); @@ -186,6 +220,8 @@ Object resolveForTask(TaskScope scope, String name) { return resolveThread(name); case REQUEST: return buildRequest(scope, name, definition); + case POOLED: + return resolvePooled(scope, name, definition); case TASK: default: break; @@ -246,6 +282,75 @@ private Object buildRequest(TaskScope scope, String name, ResourceDefinition def return value; } + /** + * Resolve a pooled resource: one checkout per task per resource, cached in the + * task scope like a task-scoped instance and returned to the pool (not + * disposed) when the task ends. + */ + private Object resolvePooled(TaskScope scope, String name, ResourceDefinition definition) { + Map cache = scope.cache(); + Object cached = cache.get(name); + if (cached != null) { + return unwrap(cached); + } + ResourcePool pool = pool(name, definition); + Object value = pool.acquire(); + cache.put(name, value == null ? NULL : value); + scope.pushTeardown(() -> pool.release(value)); + return value; + } + + /** Resolve a pooled factory's dependency, enforcing the worker-only guard. */ + private Object resolvePooledDependency(String name) { + ResourceDefinition definition = definition(name); + if (definition.scope() != ResourceScope.WORKER) { + throw new ResourceException("resource '" + name + "' is " + + scopeWord(definition.scope()) + + "-scoped; a pooled resource may only use worker resources"); + } + return resolveWorker(name); + } + + /** The pool for {@code name}, created lazily on first use. */ + private ResourcePool pool(String name, ResourceDefinition definition) { + return pools.computeIfAbsent(name, key -> createPool(name, definition)); + } + + /** + * A pool whose factory and disposer keep the per-resource counters honest: + * {@code created} moves only when the factory builds, {@code disposed} only + * when the pool actually disposes — checkout/return never touch them. + */ + private ResourcePool createPool(String name, ResourceDefinition definition) { + return new ResourcePool( + name, + definition.pool(), + () -> { + Object value = build(name, definition, pooledContext); + counter(name).created.incrementAndGet(); + return value; + }, + value -> disposePooled(name, value, definition.dispose())); + } + + /** Dispose one pooled instance; without a disposer the drop still counts as disposed. */ + private void disposePooled(String name, Object value, Consumer disposer) { + if (disposer == null) { + counter(name).disposed.incrementAndGet(); + return; + } + dispose(name, value, disposer); + } + + /** Eagerly build {@code poolMin} instances for every pooled resource that asks for prewarm. */ + private void prewarmPools() { + definitions.forEach((name, definition) -> { + if (definition.scope() == ResourceScope.POOLED && definition.pool().poolMin() > 0) { + pool(name, definition).prewarm(); + } + }); + } + /** Context handed to a request factory: dependencies resolve through the active task scope. */ private ResourceContext requestContext(TaskScope scope) { return new ResourceContext() { @@ -278,6 +383,10 @@ private Object build(String name, ResourceDefinition definition, ResourceContext } private void disposeWorker() { + // Pools first: pooled instances may depend on worker resources, which the + // teardown stack below disposes. + pools.values().forEach(ResourcePool::shutdown); + pools.clear(); synchronized (workerTeardown) { while (!workerTeardown.isEmpty()) { workerTeardown.pop().run(); diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceScope.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceScope.java index c23c5d73..afbebb18 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceScope.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/ResourceScope.java @@ -17,5 +17,12 @@ public enum ResourceScope { * Built fresh on every {@code use()} and disposed when the task ends — * never cached, so N uses inside one task yield N instances. */ - REQUEST + REQUEST, + /** + * A bounded pool of instances shared across tasks: each task checks out one + * instance for its duration and returns it at task end. Capacity is bounded + * by the {@link PoolConfig} supplied at registration; instances are disposed + * at worker shutdown or when their {@link PoolConfig#maxLifetime()} expires. + */ + POOLED } diff --git a/sdks/java/src/main/java/org/byteveda/taskito/resources/package-info.java b/sdks/java/src/main/java/org/byteveda/taskito/resources/package-info.java index 88c241db..c189e115 100644 --- a/sdks/java/src/main/java/org/byteveda/taskito/resources/package-info.java +++ b/sdks/java/src/main/java/org/byteveda/taskito/resources/package-info.java @@ -2,14 +2,17 @@ * Worker-side dependency injection: register a resource once, resolve it inside * task handlers. * - *

Four scopes: {@link org.byteveda.taskito.resources.ResourceScope#WORKER} + *

Five scopes: {@link org.byteveda.taskito.resources.ResourceScope#WORKER} * resources are built lazily once and shared across every task on the worker; * {@link org.byteveda.taskito.resources.ResourceScope#THREAD} resources are * built lazily once per worker thread and disposed at worker shutdown; * {@link org.byteveda.taskito.resources.ResourceScope#TASK} resources are built - * lazily per task invocation and disposed (LIFO) when it ends; and + * lazily per task invocation and disposed (LIFO) when it ends; * {@link org.byteveda.taskito.resources.ResourceScope#REQUEST} resources are - * built fresh on every use and disposed with the task. Handlers resolve them - * with {@link org.byteveda.taskito.resources.Resources#use(String)}. + * built fresh on every use and disposed with the task; and + * {@link org.byteveda.taskito.resources.ResourceScope#POOLED} resources live in + * a bounded pool (sized by {@link org.byteveda.taskito.resources.PoolConfig}) + * that each task checks one instance out of for its duration. Handlers resolve + * them with {@link org.byteveda.taskito.resources.Resources#use(String)}. */ package org.byteveda.taskito.resources; diff --git a/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java b/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java index 70c6d7a0..9f025e3f 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java @@ -1,19 +1,24 @@ package org.byteveda.taskito; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNotSame; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import java.nio.file.Path; +import java.time.Duration; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; import org.byteveda.taskito.errors.ResourceException; +import org.byteveda.taskito.resources.PoolConfig; +import org.byteveda.taskito.resources.ResourcePool; import org.byteveda.taskito.resources.ResourceScope; import org.byteveda.taskito.resources.ResourceStat; import org.byteveda.taskito.resources.Resources; @@ -248,4 +253,181 @@ void workerScopedResourceIsBuiltPerWorker(@TempDir Path dir) throws Exception { assertEquals(2, queue.resourceMetrics().get("perWorker").created()); } } + + @Test + @Timeout(30) + void pooledResourceReusedAcrossSequentialTasks(@TempDir Path dir) throws Exception { + int jobs = 2; + AtomicInteger disposed = new AtomicInteger(); + try (Taskito queue = + Taskito.builder().url(dir.resolve("rpl.db").toString()).open()) { + queue.resource("pooled", PoolConfig.of(1), ctx -> new Object(), value -> disposed.incrementAndGet()); + AtomicReference first = new AtomicReference<>(); + AtomicReference second = new AtomicReference<>(); + CountDownLatch ran = new CountDownLatch(jobs); + try (Worker worker = queue.worker() + .concurrency(1) + .handle(TASK, p -> { + Object instance = Resources.use("pooled"); + if (!first.compareAndSet(null, instance)) { + second.set(instance); + } + ran.countDown(); + return p; + }) + .start()) { + for (int i = 0; i < jobs; i++) { + queue.enqueue(TASK, i); + } + assertTrue(ran.await(20, TimeUnit.SECONDS)); + } // worker close shuts the pool down and disposes its instances + + assertSame(first.get(), second.get(), "a pool of one hands the same instance to sequential tasks"); + Map metrics = queue.resourceMetrics(); + assertEquals(1, metrics.get("pooled").created(), "one checkout at a time builds one instance"); + assertEquals(1, metrics.get("pooled").disposed(), "disposed at worker shutdown"); + assertEquals(1, disposed.get()); + } + } + + @Test + void pooledPoolMinCannotExceedPoolSize() { + IllegalArgumentException thrown = assertThrows( + IllegalArgumentException.class, () -> PoolConfig.of(2).withPoolMin(3)); + assertTrue(thrown.getMessage().contains("poolMin must be <= poolSize"), thrown.getMessage()); + } + + @Test + @Timeout(30) + void pooledExhaustionTimesOut() { + PoolConfig config = new PoolConfig(1, 0, Duration.ofMillis(50), null); + ResourcePool pool = new ResourcePool("db", config, Object::new, null); + assertNotNull(pool.acquire(), "first checkout takes the only permit"); + ResourceException thrown = assertThrows(ResourceException.class, pool::acquire); + assertTrue(thrown.getMessage().contains("pool timed out"), thrown.getMessage()); + assertTrue(thrown.getMessage().contains("db"), thrown.getMessage()); + assertEquals(1, pool.stats().totalTimeouts()); + } + + @Test + @Timeout(30) + void pooledFactoryFailureDoesNotLeakCapacity() { + int failures = 3; + AtomicInteger attempts = new AtomicInteger(); + Supplier factory = () -> { + if (attempts.incrementAndGet() <= failures) { + throw new IllegalStateException("connect failed"); + } + return new Object(); + }; + ResourcePool pool = + new ResourcePool("flaky", new PoolConfig(1, 0, Duration.ofMillis(200), null), factory, null); + for (int i = 0; i < failures; i++) { + assertThrows(IllegalStateException.class, pool::acquire); + assertEquals(0, pool.stats().active(), "a failed build must not count as checked out"); + } + // The permit was returned on every failure, so a size-1 pool still has capacity. + assertNotNull(pool.acquire()); + assertEquals(1, pool.stats().active()); + } + + @Test + @Timeout(30) + void pooledPrewarmBuildsMinInstances(@TempDir Path dir) throws Exception { + AtomicInteger disposed = new AtomicInteger(); + try (Taskito queue = + Taskito.builder().url(dir.resolve("rpp.db").toString()).open()) { + queue.resource( + "warm", PoolConfig.of(4).withPoolMin(2), ctx -> new Object(), value -> disposed.incrementAndGet()); + try (Worker worker = queue.worker().handle(TASK, p -> p).start()) { + assertEquals( + 2, + queue.resourceMetrics().get("warm").created(), + "prewarm builds poolMin instances at worker start, before any job"); + } + assertEquals(2, queue.resourceMetrics().get("warm").disposed()); + assertEquals(2, disposed.get()); + } + } + + @Test + @Timeout(30) + void pooledMaxLifetimeDisposesExpired() throws Exception { + AtomicInteger built = new AtomicInteger(); + AtomicInteger disposed = new AtomicInteger(); + ResourcePool pool = new ResourcePool( + "aging", + new PoolConfig(1, 0, Duration.ofSeconds(5), Duration.ofMillis(20)), + () -> "instance-" + built.incrementAndGet(), + value -> disposed.incrementAndGet()); + Object firstInstance = pool.acquire(); + pool.release(firstInstance); + Thread.sleep(60); // outlive maxLifetime while parked + Object secondInstance = pool.acquire(); + assertNotSame(firstInstance, secondInstance, "an expired idle instance is replaced, not reused"); + assertEquals(2, built.get()); + assertEquals(1, disposed.get(), "the expired instance was disposed"); + } + + @Test + @Timeout(30) + void pooledFactoryCannotUseTaskScopedResource(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("rpg.db").toString()).open()) { + queue.resource("perTask", ResourceScope.TASK, ctx -> new Object()); + queue.resource("pooled", PoolConfig.of(1), ctx -> ctx.use("perTask"), value -> {}); + AtomicReference thrown = new AtomicReference<>(); + CountDownLatch ran = new CountDownLatch(1); + try (Worker worker = queue.worker() + .handle(TASK, p -> { + try { + Resources.use("pooled"); + } catch (RuntimeException e) { + thrown.set(e); + } finally { + ran.countDown(); + } + return p; + }) + .start()) { + queue.enqueue(TASK, 1); + assertTrue(ran.await(20, TimeUnit.SECONDS)); + } + assertEquals(ResourceException.class, thrown.get().getClass()); + assertTrue( + thrown.get().getMessage().contains("a pooled resource may only use worker resources"), + thrown.get().getMessage()); + } + } + + @Test + @Timeout(30) + void pooledInstancesDisposedAtWorkerShutdown(@TempDir Path dir) throws Exception { + AtomicInteger disposed = new AtomicInteger(); + try (Taskito queue = + Taskito.builder().url(dir.resolve("rpd.db").toString()).open()) { + queue.resource("conn", PoolConfig.of(2), ctx -> new Object(), value -> disposed.incrementAndGet()); + // Hold both tasks at a barrier so each checks its own instance out. + CyclicBarrier bothBusy = new CyclicBarrier(2); + CountDownLatch ran = new CountDownLatch(2); + try (Worker worker = queue.worker() + .concurrency(2) + .handle(TASK, p -> { + Resources.use("conn"); + bothBusy.await(20, TimeUnit.SECONDS); + ran.countDown(); + return p; + }) + .start()) { + queue.enqueue(TASK, 1); + queue.enqueue(TASK, 2); + assertTrue(ran.await(25, TimeUnit.SECONDS)); + } // worker close returns both instances, then pool shutdown disposes them + + Map metrics = queue.resourceMetrics(); + assertEquals(2, metrics.get("conn").created(), "both concurrent tasks checked out an instance"); + assertEquals(2, metrics.get("conn").disposed(), "pool shutdown disposed every instance"); + assertEquals(2, disposed.get()); + } + } }