Threading refactor (4/7): Pool & op-executor contract#4
Open
igorbernstein2 wants to merge 9 commits into
Open
Conversation
igorbernstein2
force-pushed
the
threading-refactor-phase-4
branch
from
June 17, 2026 20:02
64d0864 to
bb3fd91
Compare
igorbernstein2
force-pushed
the
threading-refactor-phase-3
branch
from
June 17, 2026 20:02
8f18528 to
b32ff78
Compare
mutianf
reviewed
Jun 18, 2026
nimf
approved these changes
Jun 21, 2026
igorbernstein2
force-pushed
the
threading-refactor-phase-3
branch
from
June 22, 2026 21:41
b32ff78 to
5c4f476
Compare
igorbernstein2
force-pushed
the
threading-refactor-phase-4
branch
from
June 22, 2026 21:41
bb3fd91 to
1d18231
Compare
mutianf
approved these changes
Jun 23, 2026
Tests that call Future.get() on vRpc chains can hang indefinitely if an exception breaks the callback dispatch chain and orphans the future (see ISSUE-006). A class-level JUnit 5 @timeout(30) converts a silent hang into a clear test failure within a bounded time. Applied to: RetryingVRpcTest, VRpcTracerTest, ClientTest, TableBaseTest, SessionImplTest, SessionPoolImplTest.
PendingVRpc.cancel and drainTo move isCancelled/realCall onto ctx.getExecutor(); the pool lock now covers only queue / poolState / session list. close() switches to cancelWithResult to honor the new op-executor contract. NOOP_CALL sentinel removed. PendingVRpc.start arms the deadline monitor only AFTER committing to the queue (inside the synchronized block). Previously the timer was scheduled before the pool-state check, so the closed-pool fast-fail path returned without cancelling it — the timer then fired later and called listener.onClose a second time with DEADLINE_EXCEEDED. RetryingVRpc.Active suppressed the duplicate at the user-facing layer, but tracer.onAttemptFinish still ran twice and corrupted per-attempt metrics. Adds SessionPoolImplTest#pendingVRpcOnClosedPoolDoesNotLeakDeadlineMonitor.
VOperationImpl captures opExecutor in start() and trampolines start/cancel via it. RetryingVRpc.start runs synchronously on the op-executor task RetryingVRpc.cancel no longer wraps in execute. Tracer.onOperationStart reordered before started=true (a throwing tracer short-circuits to direct listener.onClose). listener.onMessage failures classify as USER_FAILURE. CleanupListener tracks a closed flag to prevent gRPC-context listener leaks on synchronous chain close.
…en Client.close OpExecutor switches from a SequentialExecutor backing to an internal ArrayDeque + drain loop, and gains runInline() — runs the task synchronously on the caller thread when the executor is idle, otherwise queues it. VOperationImpl uses runInline for chain.start so the start dispatch skips the queue+drain round-trip. Drain rejections (e.g. during shutdown) reset drainScheduled so subsequent submissions can retry. Client.close drains userCallbackExecutor first via a 5s-bounded shutdownAndAwait so pool.close's cancelWithResult onClose notifications complete before the executor is torn down. Resource.close becomes idempotent. Drive-by cleanups: replace fully-qualified type names with imports across touched files, and swap Guava Objects.hashCode(id) for Long.hashCode(id) in ChannelPoolDpImpl.AfeId to avoid per-call boxing and array allocation.
The tracer reaching RetryingVRpc is always CompositeVRpcTracer or NoopVrpcTracer (both produced by Metrics.newTableTracer); both guarantee no-throw: NoopVrpcTracer is empty-body, CompositeVRpcTracer wraps every child in try/catch and swallows. The defensive try/catch here and the elaborate `started=true` ordering comment justifying it were unreachable. If a future contributor wires in a tracer that bypasses Composite, the OpExecutor uncaught handler installed by VOperationImpl.start is the remaining safety net.
…n paths
Previously the cancelled-pending-drain path in SessionPoolImpl synthesized a
Duration.ZERO + Status.CANCELLED result and routed through onVRpcFinish,
relying on its internal isOk() gate to skip the picker-latency update. The
sentinel value at the call site was misleading and forced readers to chase
through onVRpcFinish to confirm nothing was actually being measured.
Split SessionHandle.onVRpcFinish into:
- onVRpcFinish(elapsed, result): vRPC reached the wire; updates picker
latency on success as before.
- onPendingVRpcCancelled(): vRPC was cancelled before attaching to a real
call; pool-occupancy bookkeeping only.
Both share a private releaseToPool() helper. SessionPoolImpl mirrors the
split with onVRpcComplete + onPendingVRpcCancelled, factoring the post-
release tryDrain logic into afterRelease (annotated @GuardedBy("this")).
Each call site now names what happened; no more Duration.ZERO sentinel and
no more synthesized CANCELLED result. Behaviorally identical: the only
intra-method ordering change (latency update happens after the AFE re-add
inside onVRpcFinish) is invisible under the pool's synchronized (this).
Adds a SessionListTest case asserting that onPendingVRpcCancelled performs
the same bookkeeping as the OK path but does not touch the AFE's latency
stats.
igorbernstein2
force-pushed
the
threading-refactor-phase-3
branch
from
June 29, 2026 17:20
5c4f476 to
e4ca57f
Compare
igorbernstein2
force-pushed
the
threading-refactor-phase-4
branch
from
June 29, 2026 17:20
1d18231 to
d824fd4
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Phase 4 of 7. Routes
PendingVRpcper-op state through the op executor, consolidates cancel trampolines atVOperationImpl, and replaces theOpExecutorbacking with an inline-capable queue.Test infrastructure first: a 30-second
@Timeouton the session/pool tests so a callback-dispatch regression surfaces as a clear failure instead of a silent hang.What's in here
test: add 30-second @Timeout to all session/pool integration tests@Timeout(30)onRetryingVRpcTest,VRpcTracerTest,ClientTest,TableBaseTest,SessionImplTest,SessionPoolImplTest.chore: route PendingVRpc per-op state through the op executor(+ deadline-monitor fix)PendingVRpc.cancelanddrainTomoveisCancelled/realCallontoctx.getExecutor(); the pool lock now covers only queue /PendingVRpc.startarms the deadline monitor only AFTER committing to the queue (previously a closed-pool fast-fail would leak the timer, which then fired and corrupted tracerchore: consolidate cancel trampolines at VOperationImplVOperationImplcapturesopExecutorinstart()and trampolines start/cancel via it.RetryingVRpc.startruns synchronously on the op-executor task.tracer.onOperationStartreordered beforestarted=true(a throwing tracer short-circuits to directlistener.onClose).listener.onMessagefailures classify asUSER_FAILURE.chore: replace OpExecutor backing with an inline-capable queue; tighten Client.close(+ FQ-name cleanup)OpExecutorswitches fromSequentialExecutorto an internalArrayDeque+ drain loop, and gainsrunInline()— runs synchronously on the caller when idle, otherwise queues.VOperationImplusesrunInlineforchain.startso the start dispatch skips the queue+drain round-trip.Client.closedrainsuserCallbackExecutorfirst via a 5s-boundedshutdownAndAwait. Drive-by: replace fully-qualified type names with imports; swap GuavaObjects.hashCode(id)forLong.hashCode(id)inChannelPoolDpImpl.AfeId.Stack position
threading-refactor-phase-3threading-refactor-phase-5— close ordering & shutdown hardening