From 80d372f2fd93bd885ea84bbf294abd59696848b6 Mon Sep 17 00:00:00 2001 From: Pratyush Sharma <56130065+pratyush618@users.noreply.github.com> Date: Mon, 6 Jul 2026 10:45:03 +0530 Subject: [PATCH] feat(java): add THREAD and REQUEST resource scopes --- .../taskito/resources/ResourceContext.java | 8 +- .../taskito/resources/ResourceRuntime.java | 105 +++++++++++++++- .../taskito/resources/ResourceScope.java | 14 ++- .../taskito/resources/package-info.java | 10 +- .../org/byteveda/taskito/ResourceTest.java | 119 ++++++++++++++++++ 5 files changed, 245 insertions(+), 11 deletions(-) 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 70633516..f18119e8 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 @@ -1,9 +1,11 @@ package org.byteveda.taskito.resources; /** - * Handed to a resource factory so it can depend on other resources. A - * {@code WORKER} factory may only {@link #use} other worker resources; a - * {@code TASK} factory may use either. + * 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 + * worker or thread resources, and {@code TASK}/{@code REQUEST} factories may + * use any scope. */ public interface ResourceContext { /** The scope the factory is building for. */ 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 19e4dfdd..030cc188 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 @@ -6,6 +6,7 @@ import java.util.Deque; import java.util.LinkedHashMap; import java.util.LinkedHashSet; +import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -34,6 +35,13 @@ public final class ResourceRuntime { private final ConcurrentMap counters; private final ConcurrentMap workerCache = new ConcurrentHashMap<>(); private final ConcurrentMap workerLocks = new ConcurrentHashMap<>(); + /** + * Per-name, per-thread instances for {@code THREAD}-scoped resources. Only the + * owning thread reads or writes its own entry (so no per-name lock is needed); + * the concurrent maps exist for cross-thread visibility at worker teardown. + */ + private final ConcurrentMap> threadCache = new ConcurrentHashMap<>(); + private final Deque workerTeardown = new ArrayDeque<>(); /** Names being resolved on the current thread, so a dependency cycle fails fast instead of recursing. */ private final ThreadLocal> resolving = ThreadLocal.withInitial(LinkedHashSet::new); @@ -53,6 +61,19 @@ public T use(String name) { } }; + /** Context handed to a thread factory: it may use worker or thread resources. */ + private final ResourceContext threadContext = new ResourceContext() { + @Override + public ResourceScope scope() { + return ResourceScope.THREAD; + } + + @Override + public T use(String name) { + return cast(resolveThread(name)); + } + }; + /** A client-level runtime: holds definitions + counters, hands each worker a child via {@link #forWorker()}. */ public ResourceRuntime() { this.definitions = new ConcurrentHashMap<>(); @@ -122,7 +143,9 @@ public Map metrics() { Object resolveWorker(String name) { ResourceDefinition definition = definition(name); if (definition.scope() != ResourceScope.WORKER) { - throw new ResourceException("resource '" + name + "' is task-scoped; a worker resource cannot use it"); + throw new ResourceException("resource '" + name + "' is " + + scopeWord(definition.scope()) + + "-scoped; a worker resource may only use worker resources"); } Object cached = workerCache.get(name); if (cached != null) { @@ -149,11 +172,23 @@ Object resolveWorker(String name) { } } - /** Resolve for a task: worker-scoped hits the shared cache; task-scoped builds once per invocation. */ + /** + * 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. + */ Object resolveForTask(TaskScope scope, String name) { ResourceDefinition definition = definition(name); - if (definition.scope() == ResourceScope.WORKER) { - return resolveWorker(name); + switch (definition.scope()) { + case WORKER: + return resolveWorker(name); + case THREAD: + return resolveThread(name); + case REQUEST: + return buildRequest(scope, name, definition); + case TASK: + default: + break; } Map cache = scope.cache(); Object cached = cache.get(name); @@ -169,6 +204,63 @@ Object resolveForTask(TaskScope scope, String name) { return value; } + /** + * Resolve a thread-scoped resource for the current worker thread, building it + * lazily. Only the owning thread touches its own map entry, so a plain + * get-build-put is race-free; the shared {@code workerTeardown} deque keeps + * disposal globally LIFO across worker and thread instances. + */ + Object resolveThread(String name) { + ResourceDefinition definition = definition(name); + if (definition.scope() == ResourceScope.WORKER) { + return resolveWorker(name); + } + if (definition.scope() != ResourceScope.THREAD) { + throw new ResourceException("resource '" + name + "' is " + + scopeWord(definition.scope()) + + "-scoped; a thread resource may only use worker or thread resources"); + } + ConcurrentMap perThread = threadCache.computeIfAbsent(name, key -> new ConcurrentHashMap<>()); + Object cached = perThread.get(Thread.currentThread()); + if (cached != null) { + return unwrap(cached); + } + Object value = build(name, definition, threadContext); + perThread.put(Thread.currentThread(), value == null ? NULL : value); + counter(name).created.incrementAndGet(); + if (definition.dispose() != null) { + synchronized (workerTeardown) { + workerTeardown.push(() -> dispose(name, value, definition.dispose())); + } + } + return value; + } + + /** Build a request-scoped resource: fresh on every use, disposed with the task (LIFO). */ + private Object buildRequest(TaskScope scope, String name, ResourceDefinition definition) { + Object value = build(name, definition, requestContext(scope)); + counter(name).created.incrementAndGet(); + if (definition.dispose() != null) { + scope.pushTeardown(() -> dispose(name, value, definition.dispose())); + } + return value; + } + + /** Context handed to a request factory: dependencies resolve through the active task scope. */ + private ResourceContext requestContext(TaskScope scope) { + return new ResourceContext() { + @Override + public ResourceScope scope() { + return ResourceScope.REQUEST; + } + + @Override + public T use(String name) { + return scope.use(name); + } + }; + } + private Object build(String name, ResourceDefinition definition, ResourceContext context) { Set chain = resolving.get(); if (!chain.add(name)) { @@ -192,6 +284,11 @@ private void disposeWorker() { } } workerCache.clear(); + threadCache.clear(); + } + + private static String scopeWord(ResourceScope scope) { + return scope.name().toLowerCase(Locale.ROOT); } private void dispose(String name, Object value, Consumer disposer) { 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 973f1cf0..c23c5d73 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 @@ -4,6 +4,18 @@ public enum ResourceScope { /** Built once, lazily, and shared across every task on the worker. */ WORKER, + /** + * Built lazily once per worker thread and shared by every task that runs on + * that thread; disposed when the worker shuts down. If the pool retires a + * thread early (cached pools, autoscale), its instance is retained until + * worker teardown — bounded by the number of threads the pool ever created. + */ + THREAD, /** Built lazily per task invocation and disposed when the task ends. */ - TASK + TASK, + /** + * Built fresh on every {@code use()} and disposed when the task ends — + * never cached, so N uses inside one task yield N instances. + */ + REQUEST } 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 ae4a6bca..88c241db 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,10 +2,14 @@ * Worker-side dependency injection: register a resource once, resolve it inside * task handlers. * - *

Two scopes (mirroring the Node SDK): {@link org.byteveda.taskito.resources.ResourceScope#WORKER} + *

Four 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. Handlers resolve - * them with {@link org.byteveda.taskito.resources.Resources#use(String)}. + * lazily per task invocation and disposed (LIFO) when it ends; and + * {@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)}. */ 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 1bb08715..70c6d7a0 100644 --- a/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java +++ b/sdks/java/src/test/java/org/byteveda/taskito/ResourceTest.java @@ -1,6 +1,8 @@ package org.byteveda.taskito; import static org.junit.jupiter.api.Assertions.assertEquals; +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; @@ -103,6 +105,123 @@ void circularResourceDependencyIsRejected(@TempDir Path dir) throws Exception { } } + @Test + @Timeout(30) + void threadResourceReusedAcrossSequentialTasks(@TempDir Path dir) throws Exception { + int jobs = 2; + AtomicInteger disposed = new AtomicInteger(); + try (Taskito queue = + Taskito.builder().url(dir.resolve("rth.db").toString()).open()) { + queue.resource("perThread", ResourceScope.THREAD, 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) // one thread → both jobs share its instance + .handle(TASK, p -> { + Object instance = Resources.use("perThread"); + 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 disposes thread-scoped instances + + assertSame(first.get(), second.get(), "same thread reuses its instance"); + assertEquals(1, queue.resourceMetrics().get("perThread").created()); + assertEquals(1, disposed.get(), "disposed once at worker shutdown"); + } + } + + @Test + @Timeout(30) + void requestResourceFreshPerUse(@TempDir Path dir) throws Exception { + AtomicInteger disposed = new AtomicInteger(); + try (Taskito queue = + Taskito.builder().url(dir.resolve("rrq.db").toString()).open()) { + queue.resource("perUse", ResourceScope.REQUEST, ctx -> new Object(), value -> disposed.incrementAndGet()); + AtomicReference a = new AtomicReference<>(); + AtomicReference b = new AtomicReference<>(); + CountDownLatch ran = new CountDownLatch(1); + try (Worker worker = queue.worker() + .handle(TASK, p -> { + a.set(Resources.use("perUse")); + b.set(Resources.use("perUse")); + ran.countDown(); + return p; + }) + .start()) { + queue.enqueue(TASK, 1); + assertTrue(ran.await(20, TimeUnit.SECONDS)); + } + assertNotSame(a.get(), b.get(), "every use builds a fresh instance"); + assertEquals(2, queue.resourceMetrics().get("perUse").created()); + assertEquals(2, disposed.get(), "both instances disposed at task end"); + } + } + + @Test + @Timeout(30) + void threadFactoryCannotUseTaskScopedResource(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("rtg.db").toString()).open()) { + queue.resource("perTask", ResourceScope.TASK, ctx -> new Object()); + queue.resource("perThread", ResourceScope.THREAD, ctx -> ctx.use("perTask")); + AtomicReference> thrown = new AtomicReference<>(); + CountDownLatch ran = new CountDownLatch(1); + try (Worker worker = queue.worker() + .handle(TASK, p -> { + try { + Resources.use("perThread"); + } catch (RuntimeException e) { + thrown.set(e.getClass()); + } finally { + ran.countDown(); + } + return p; + }) + .start()) { + queue.enqueue(TASK, 1); + assertTrue(ran.await(20, TimeUnit.SECONDS)); + } + assertEquals(ResourceException.class, thrown.get()); + } + } + + @Test + @Timeout(30) + void workerFactoryCannotUseThreadResource(@TempDir Path dir) throws Exception { + try (Taskito queue = + Taskito.builder().url(dir.resolve("rwg.db").toString()).open()) { + queue.resource("perThread", ResourceScope.THREAD, ctx -> new Object()); + queue.resource("shared", ctx -> ctx.use("perThread")); // WORKER factory + AtomicReference> thrown = new AtomicReference<>(); + CountDownLatch ran = new CountDownLatch(1); + try (Worker worker = queue.worker() + .handle(TASK, p -> { + try { + Resources.use("shared"); + } catch (RuntimeException e) { + thrown.set(e.getClass()); + } finally { + ran.countDown(); + } + return p; + }) + .start()) { + queue.enqueue(TASK, 1); + assertTrue(ran.await(20, TimeUnit.SECONDS)); + } + assertEquals(ResourceException.class, thrown.get()); + } + } + @Test @Timeout(30) void workerScopedResourceIsBuiltPerWorker(@TempDir Path dir) throws Exception {