diff --git a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/session/SessionPoolImpl.java b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/session/SessionPoolImpl.java index 50542c7f5fdb..7046804e52a5 100644 --- a/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/session/SessionPoolImpl.java +++ b/java-bigtable/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/internal/session/SessionPoolImpl.java @@ -69,6 +69,8 @@ import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import java.util.logging.Level; import java.util.logging.Logger; import javax.annotation.Nullable; @@ -86,6 +88,25 @@ private enum PoolState { private static final int PROTOCOL_VERSION = 0; + // Pool-wide lock. Replaces the previous `synchronized (this)` monitor. Kept as an explicit + // ReentrantLock so the request-serving hot path can acquire it with a bounded timeout + // (tryLock) instead of blocking forever. In production we observed a pool whose monitor had + // no owner but many waiters (an orphaned monitor, e.g. after a thread died mid-critical-section + // during native-thread OOM): every incoming RPC probe (hasSession/newCall) blocked on it + // permanently, which piled up threads and shed 100% of traffic. With a ReentrantLock, a wedged + // pool degrades to a bounded fast-fail on the request path rather than an unbounded hang. + // + // Reentrancy note: several internal callbacks (createSession, onSessionGoAway, ...) acquire the + // lock while an outer critical section already holds it. ReentrantLock preserves the reentrant + // semantics the old `synchronized` relied on. + private final ReentrantLock poolLock = new ReentrantLock(); + + // How long the request-serving hot path (hasSession/newCall/PendingVRpc.start) waits to acquire + // the pool lock before giving up and degrading. Healthy critical sections are microseconds long, + // so this is only ever exhausted when the pool is wedged; keeping it bounded turns a permanent + // hang into a recoverable fast-fail. + private static final long HOT_PATH_LOCK_TIMEOUT_MILLIS = 1000; + private final Metrics metrics; private final FeatureFlags featureFlags; private final SessionPoolInfo info; @@ -96,21 +117,21 @@ private enum PoolState { // Set once by start(), and read by both user & grpc threads private volatile OpenParams openParams; - @GuardedBy("this") + @GuardedBy("poolLock") private PoolState poolState = PoolState.NEW; @VisibleForTesting - @GuardedBy("this") + @GuardedBy("poolLock") final SessionList sessions; - @GuardedBy("this") + @GuardedBy("poolLock") private final DynamicPicker picker; - @GuardedBy("this") + @GuardedBy("poolLock") private final PoolSizer poolSizer; // TODO: we need to close pendingVRpcs when deadline expires - @GuardedBy("this") + @GuardedBy("poolLock") private final Deque> pendingRpcs = new ArrayDeque<>(); private final Watchdog watchdog; @@ -124,7 +145,7 @@ private enum PoolState { // run pool-lock-holding work inline. private final Executor backgroundExecutor; - @GuardedBy("this") + @GuardedBy("poolLock") private int consecutiveFailures = 0; /** @@ -135,11 +156,11 @@ private enum PoolState { private volatile int consecutiveUnimplementedFailures = 0; // Self-rescheduling AFE-prune chain. Set by scheduleNextAfePrune; cancelled by close. - @GuardedBy("this") + @GuardedBy("poolLock") @Nullable private BigtableTimer.Timeout afeListPruneTimeout; - @GuardedBy("this") + @GuardedBy("poolLock") private boolean closed = false; // Completed when this pool has been close()d AND every session has reached the CLOSED terminal @@ -147,11 +168,11 @@ private enum PoolState { // onto userCallbackExecutor before that executor is shut down. private final CompletableFuture drainedFuture = new CompletableFuture<>(); - @GuardedBy("this") + @GuardedBy("poolLock") private BigtableTimer.Timeout retryCreateSessionFuture = null; // TODO: get the max from ClientConfiguration - @GuardedBy("this") + @GuardedBy("poolLock") private final SessionCreationBudget budget; private final ClientConfigurationManager configManager; @@ -159,7 +180,7 @@ private enum PoolState { private final DebugTagTracer debugTagTracer; - // @SuppressWarnings("GuardedBy"): error-prone flags writes to @GuardedBy("this") fields + // @SuppressWarnings("GuardedBy"): error-prone flags writes to @GuardedBy("poolLock") fields // (sessions, picker, poolSizer, pendingRpcs, budget, retryCreateSessionFuture) inside the // constructor without holding the monitor. This is safe because the object is not yet published // to other threads — no external reference exists until the constructor returns. @@ -237,7 +258,7 @@ public SessionPoolImpl( // Watchdog checks for sessions in WAIT_SERVER_CLOSE state and runs every 5 minutes watchdog = new Watchdog( - this, timer, backgroundExecutor, Duration.ofMinutes(5), sessions, debugTagTracer); + poolLock, timer, backgroundExecutor, Duration.ofMinutes(5), sessions, debugTagTracer); // Heartbeat monitoring is now done per-session via SessionImpl.scheduleHeartbeatCheck. scheduleNextAfePrune(); @@ -248,15 +269,18 @@ public SessionPoolImpl( configManager.addListener( clientConfig -> clientConfig.getSessionConfiguration().getSessionPoolConfiguration(), newConfig -> { - synchronized (SessionPoolImpl.this) { + poolLock.lock(); + try { budget.updateConfig(newConfig); poolSizer.updateConfig(newConfig); picker.updateConfig(newConfig); + } finally { + poolLock.unlock(); } }); } - @GuardedBy("this") + @GuardedBy("poolLock") private void scheduleNextAfePrune() { if (closed) { return; @@ -270,7 +294,8 @@ private void scheduleNextAfePrune() { } private void runAfePruneAndReschedule() { - synchronized (SessionPoolImpl.this) { + poolLock.lock(); + try { try { if (closed) { return; @@ -281,6 +306,8 @@ private void runAfePruneAndReschedule() { } finally { scheduleNextAfePrune(); } + } finally { + poolLock.unlock(); } } @@ -294,7 +321,8 @@ public void close(CloseSessionRequest req) { configListenerHandle.close(); List> toCancel; - synchronized (this) { + poolLock.lock(); + try { if (poolState == PoolState.CLOSED) { logger.fine(String.format("Tried to close a closed SessionPool %s", info.getLogName())); return; @@ -321,6 +349,8 @@ public void close(CloseSessionRequest req) { if (sessions.getAllSessions().isEmpty()) { drainedFuture.complete(null); } + } finally { + poolLock.unlock(); } // cancelWithResult trampolines through ctx.getExecutor() — required because the public @@ -354,31 +384,37 @@ public boolean awaitTerminated(Duration timeout) throws InterruptedException { } @Override - public synchronized void start(OpenReqT openReq, Metadata md) { - Preconditions.checkState(poolState == PoolState.NEW); - - Metadata mergedMd = new Metadata(); - mergedMd.merge(md); - mergedMd.merge(Util.composeMetadata(featureFlags, descriptor.extractHeaderParams(openReq))); - - openParams = - OpenParams.create( - mergedMd, - OpenSessionRequest.newBuilder() - .setProtocolVersion(PROTOCOL_VERSION) - .setFlags(featureFlags) - .setConsecutiveFailedConnectionAttempts(0) // will be updated each handshake attempt - .setRoutingCookie(ByteString.EMPTY) // set when each session is renegotiated - .setPayload(openReq.toByteString()) // will be set on start - .build()); - poolState = PoolState.STARTED; // TODO: maybe need a READY state as well? - - // Pre-start - for (int i = poolSizer.getScaleDelta(); i > 0; i--) { - createSession(openParams); - } - - watchdog.start(); + public void start(OpenReqT openReq, Metadata md) { + poolLock.lock(); + try { + Preconditions.checkState(poolState == PoolState.NEW); + + Metadata mergedMd = new Metadata(); + mergedMd.merge(md); + mergedMd.merge(Util.composeMetadata(featureFlags, descriptor.extractHeaderParams(openReq))); + + openParams = + OpenParams.create( + mergedMd, + OpenSessionRequest.newBuilder() + .setProtocolVersion(PROTOCOL_VERSION) + .setFlags(featureFlags) + .setConsecutiveFailedConnectionAttempts( + 0) // will be updated each handshake attempt + .setRoutingCookie(ByteString.EMPTY) // set when each session is renegotiated + .setPayload(openReq.toByteString()) // will be set on start + .build()); + poolState = PoolState.STARTED; // TODO: maybe need a READY state as well? + + // Pre-start + for (int i = poolSizer.getScaleDelta(); i > 0; i--) { + createSession(openParams); + } + + watchdog.start(); + } finally { + poolLock.unlock(); + } } private static SessionCreationBudget createInitialBudget( @@ -400,7 +436,8 @@ private int getMaxConsecutiveFailures(ClientConfigurationManager configManager) .getConsecutiveSessionFailureThreshold(); } - private synchronized void createSession(OpenParams openParams) { + @GuardedBy("poolLock") + private void createSession(OpenParams openParams) { if (!budget.tryReserveSession()) { debugTagTracer.record(TelemetryConfiguration.Level.WARN, "session_pool_no_budget"); logger.fine( @@ -452,7 +489,7 @@ public void onClose(SessionState prevState, Status status, Metadata trailers) { } } - @GuardedBy("this") + @GuardedBy("poolLock") private void maybeScheduleCreateSessionRetry() { if (retryCreateSessionFuture != null) { return; @@ -467,11 +504,14 @@ private void maybeScheduleCreateSessionRetry() { retryCreateSessionFuture = timer.newTimeout( () -> { - synchronized (SessionPoolImpl.this) { + poolLock.lock(); + try { retryCreateSessionFuture = null; if (poolState != PoolState.CLOSED && poolSizer.getScaleDelta() > 0) { createSession(openParams); } + } finally { + poolLock.unlock(); } }, backgroundExecutor, @@ -479,45 +519,59 @@ private void maybeScheduleCreateSessionRetry() { TimeUnit.MILLISECONDS); } - private synchronized void onSessionReady(SessionHandle handle, OpenSessionResponse ignored) { - logger.log( - Level.FINE, - "Session {0} in state {1}", - new Object[] {handle.getSession().getLogName(), handle.getSession().getState()}); + private void onSessionReady(SessionHandle handle, OpenSessionResponse ignored) { + poolLock.lock(); + try { + logger.log( + Level.FINE, + "Session {0} in state {1}", + new Object[] {handle.getSession().getLogName(), handle.getSession().getState()}); - consecutiveFailures = 0; - consecutiveUnimplementedFailures = 0; + consecutiveFailures = 0; + consecutiveUnimplementedFailures = 0; - if (poolState != PoolState.STARTED) { - logger.fine( - String.format( - "%s Session became ready after pool transitioned to %s, ignoring", - handle.getSession().getLogName(), poolState)); - // The session will be closed as part of SessionList#close - return; - } - handle.onSessionStarted(); + if (poolState != PoolState.STARTED) { + logger.fine( + String.format( + "%s Session became ready after pool transitioned to %s, ignoring", + handle.getSession().getLogName(), poolState)); + // The session will be closed as part of SessionList#close + return; + } + handle.onSessionStarted(); - budget.onSessionCreationSuccess(); + budget.onSessionCreationSuccess(); - // handle pending rpcs - tryDrainPendingRpcs(); + // handle pending rpcs + tryDrainPendingRpcs(); + } finally { + poolLock.unlock(); + } } - private synchronized void onVRpcComplete( - SessionHandle handle, Duration elapsed, VRpcResult result) { - handle.onVRpcFinish(elapsed, result); - afterRelease(handle); + private void onVRpcComplete(SessionHandle handle, Duration elapsed, VRpcResult result) { + poolLock.lock(); + try { + handle.onVRpcFinish(elapsed, result); + afterRelease(handle); + } finally { + poolLock.unlock(); + } } // Called when a pending vRPC was drained but cancelled before it attached to a real call. // Mirrors onVRpcComplete; the handle reports no latency because nothing ran on it. - private synchronized void onPendingVRpcCancelled(SessionHandle handle) { - handle.onPendingVRpcCancelled(); - afterRelease(handle); + private void onPendingVRpcCancelled(SessionHandle handle) { + poolLock.lock(); + try { + handle.onPendingVRpcCancelled(); + afterRelease(handle); + } finally { + poolLock.unlock(); + } } - @GuardedBy("this") + @GuardedBy("poolLock") private void afterRelease(SessionHandle handle) { // pool is shutting down, dont try to drain vrpcs if (poolState != PoolState.STARTED) { @@ -530,20 +584,25 @@ private void afterRelease(SessionHandle handle) { tryDrainPendingRpcs(); } - private synchronized void onSessionGoAway(SessionHandle handle, GoAwayResponse msg) { - handle.onSessionClosing(); - - // If the session received refresh config or pool requires a replacement, keep the current - // session and request a replacement. - // TODO: remove the check on if the open params is updated. In the future, server should - // broadcast session refresh config to all the sessions that needs reconnect and client - // just need to recreate sessions when pool sizer requries a replacement. - if (handle.getSession().isOpenParamsUpdated() || poolSizer.handleGoAway(msg)) { - logger.fine( - String.format( - "Adding new session to replace a going away session %s", - handle.getSession().getLogName())); - createSession(handle.getSession().getOpenParams()); + private void onSessionGoAway(SessionHandle handle, GoAwayResponse msg) { + poolLock.lock(); + try { + handle.onSessionClosing(); + + // If the session received refresh config or pool requires a replacement, keep the current + // session and request a replacement. + // TODO: remove the check on if the open params is updated. In the future, server should + // broadcast session refresh config to all the sessions that needs reconnect and client + // just need to recreate sessions when pool sizer requries a replacement. + if (handle.getSession().isOpenParamsUpdated() || poolSizer.handleGoAway(msg)) { + logger.fine( + String.format( + "Adding new session to replace a going away session %s", + handle.getSession().getLogName())); + createSession(handle.getSession().getOpenParams()); + } + } finally { + poolLock.unlock(); } } @@ -552,7 +611,8 @@ private void onSessionClose( List> toBeClosed = new ArrayList<>(); - synchronized (this) { + poolLock.lock(); + try { logger.fine( String.format("Removing closed session from pool %s", handle.getSession().getLogName())); @@ -594,6 +654,8 @@ private void onSessionClose( createSession(openParams.withIncrementedAttempts()); } } + } finally { + poolLock.unlock(); } if (!toBeClosed.isEmpty()) { @@ -616,7 +678,7 @@ private void onSessionClose( } } - @GuardedBy("this") + @GuardedBy("poolLock") private void tryDrainPendingRpcs() { while (!pendingRpcs.isEmpty()) { Optional handle = picker.pickSession(); @@ -628,7 +690,7 @@ private void tryDrainPendingRpcs() { } } - @GuardedBy("this") + @GuardedBy("poolLock") private List> popClosableRpcs() { List> toBeClosed = new ArrayList<>(); Iterator> iter = pendingRpcs.iterator(); @@ -641,19 +703,57 @@ private void tryDrainPendingRpcs() { } @Override - public synchronized VRpc newCall( + // @SuppressWarnings("GuardedBy"): error-prone can't see that the guarded fields are only touched + // inside the tryLock()-guarded region. Bounded acquisition on the request hot path; see poolLock. + @SuppressWarnings("GuardedBy") + public VRpc newCall( VRpcDescriptor desc) { - Optional handle = picker.pickSession(); - if (handle.isPresent()) { - return newRealCall(desc, handle.get()); + if (!tryAcquireHotPathLock()) { + // The pool lock is wedged. Degrade to a pending call instead of blocking the caller forever; + // PendingVRpc.start() will re-attempt the bounded acquisition and fast-fail if still wedged. + return new PendingVRpc<>(desc); } - if (logger.isLoggable(Level.FINE)) { - logger.fine( - String.format( - "%s Creating new rpc as pending, numPending: %d, %s", - info.getLogName(), pendingRpcs.size(), sessions.getStats())); + try { + Optional handle = picker.pickSession(); + if (handle.isPresent()) { + return newRealCall(desc, handle.get()); + } + if (logger.isLoggable(Level.FINE)) { + logger.fine( + String.format( + "%s Creating new rpc as pending, numPending: %d, %s", + info.getLogName(), pendingRpcs.size(), sessions.getStats())); + } + return new PendingVRpc<>(desc); + } finally { + poolLock.unlock(); + } + } + + // Acquire the pool lock with a bounded timeout for the request-serving hot path. Returns false if + // the lock could not be acquired within HOT_PATH_LOCK_TIMEOUT_MILLIS (i.e. the pool is wedged) or + // the caller was interrupted. Callers that get false MUST degrade instead of accessing guarded + // state, and MUST NOT call poolLock.unlock(). + private boolean tryAcquireHotPathLock() { + // Fast path: the uncontended lock is acquired immediately and, unlike the timed tryLock below, + // WITHOUT throwing if the calling thread already has its interrupt flag set. This preserves the + // old `synchronized` behavior for the common healthy case — a pre-existing interrupt (e.g. on a + // reused worker thread) must not be mistaken for a wedged pool, which would spuriously shed + // traffic and, via the interrupt re-assertion below, poison every later call on that thread. + if (poolLock.tryLock()) { + return true; } - return new PendingVRpc<>(desc); + boolean locked = false; + try { + locked = poolLock.tryLock(HOT_PATH_LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + // Preserve interrupt status for the caller; treat as a failed (degraded) acquisition. + Thread.currentThread().interrupt(); + } + if (!locked) { + debugTagTracer.record(TelemetryConfiguration.Level.ERROR, "session_pool_lock_timeout"); + } + return locked; } private VRpc newRealCall( @@ -686,8 +786,22 @@ public int getConsecutiveUnimplementedFailures() { // Used in the shim layer for fallback decisions. Called on every RPC. @Override - public synchronized boolean hasSession() { - return sessions.getStats().getReadyCount() + sessions.getStats().getInUseCount() > 0; + // @SuppressWarnings("GuardedBy"): sessions is only read inside the tryLock()-guarded region. + @SuppressWarnings("GuardedBy") + public boolean hasSession() { + if (!tryAcquireHotPathLock()) { + // Treat a wedged pool as "no session available" rather than blocking on the pool lock + // forever. + // Note: the shim's routing OR (unimplementedFailures < MAX || hasSession()) means returning + // false here does not by itself force the classic path; the wedge protection that matters is + // newCall()/PendingVRpc.start() fast-failing with a retriable UNAVAILABLE. + return false; + } + try { + return sessions.getStats().getReadyCount() + sessions.getStats().getInUseCount() > 0; + } finally { + poolLock.unlock(); + } } class PendingVRpc implements VRpc { @@ -707,6 +821,9 @@ public PendingVRpc(VRpcDescriptor desc) { } @Override + // @SuppressWarnings("GuardedBy"): guarded state is only touched inside the tryLock()-guarded + // region. Bounded acquisition on the request hot path; see poolLock. + @SuppressWarnings("GuardedBy") public void start(ReqT req, VRpcCallContext ctx, VRpcListener listener) { Preconditions.checkState(this.req == null, "request is already started"); Preconditions.checkNotNull(req, "request can't be null"); @@ -717,7 +834,25 @@ public void start(ReqT req, VRpcCallContext ctx, VRpcListener listener) { this.ctx = ctx; this.listener = listener; - synchronized (SessionPoolImpl.this) { + if (!tryAcquireHotPathLock()) { + // The pool lock is wedged. Fast-fail this call with UNAVAILABLE (uncommitted, so it stays + // retriable) instead of blocking the caller thread forever. Mark it terminal inside the + // op-executor (which owns isCancelled) so a subsequent cancel() is a no-op rather than + // delivering a second onClose. + VRpcResult result = + VRpcResult.createUncommitedError( + Status.UNAVAILABLE.withCause( + new IllegalStateException("SessionPool lock unavailable (pool wedged)"))); + ctx.getExecutor() + .execute( + () -> { + if (isCancelled) return; + isCancelled = true; + listener.onClose(result); + }); + return; + } + try { if (SessionPoolImpl.this.poolState != PoolState.STARTED) { VRpcResult result = VRpcResult.createUncommitedError( @@ -745,6 +880,8 @@ public void start(ReqT req, VRpcCallContext ctx, VRpcListener listener) { } tryDrainPendingRpcs(); + } finally { + poolLock.unlock(); } } @@ -762,9 +899,25 @@ public void cancel(@Nullable String message, @Nullable Throwable cause) { // cancel() and drainTo() are sequenced via ctx.getExecutor() (a per-op SerializingExecutor), // so isCancelled and realCall are owned exclusively by that executor — no pool lock needed. + // @SuppressWarnings("GuardedBy"): pendingRpcs is only touched inside the tryLock()-guarded + // region below. + @SuppressWarnings("GuardedBy") private void cancel(Status status, boolean onlyCancelPendingCall) { - synchronized (SessionPoolImpl.this) { - pendingRpcs.remove(this); // eager removal; no-op if already drained + // The eager removal is only an optimization to skip a wasted session pick in + // tryDrainPendingRpcs. + // Use the bounded hot-path acquisition instead of a blocking lock(): this runs on a callback + // (op-executor) thread, and an orphaned/wedged poolLock would otherwise park that thread + // forever + // — the exact thread pile-up that wedged pods in production. If we can't get the lock, skip + // the + // removal; drainTo()'s isCancelled guard is the correctness backstop (it releases the session + // via onPendingVRpcCancelled instead of starting the real call). See poolLock. + if (tryAcquireHotPathLock()) { + try { + pendingRpcs.remove(this); // eager removal; no-op if already drained + } finally { + poolLock.unlock(); + } } ctx.getExecutor() .execute( @@ -844,7 +997,7 @@ private BigtableTimer.Timeout monitorDeadline(Deadline deadline) { static class Watchdog implements Runnable { private static final Logger LOG = Logger.getLogger(Watchdog.class.getName()); - private final Object lock; + private final Lock lock; private final BigtableTimer timer; private final Executor backgroundExecutor; private final Duration interval; @@ -862,12 +1015,9 @@ static class Watchdog implements Runnable { @GuardedBy("scheduleLock") private boolean watchdogClosed = false; - // The `lock` parameter is the pool-wide monitor (SessionPoolImpl.this). It is typed as Object - // because Watchdog is a static nested class and cannot reference the outer instance type in its - // constructor signature without creating a circular dependency. Phase 5 will replace this with - // a properly typed lock once the per-AFE sharding model is established. + // The `lock` parameter is the pool-wide lock (SessionPoolImpl.poolLock). public Watchdog( - Object lock, + Lock lock, BigtableTimer timer, Executor backgroundExecutor, Duration interval, @@ -885,7 +1035,7 @@ public Watchdog( @VisibleForTesting Watchdog( - Object lock, + Lock lock, BigtableTimer timer, Executor backgroundExecutor, Duration interval, @@ -938,8 +1088,11 @@ public void run() { // ConcurrentModificationException. Most common trigger is a heartbeat-miss cascade that // churns sessions while the watchdog is walking the list. Set allSessions; - synchronized (lock) { + lock.lock(); + try { allSessions = new HashSet<>(sessions.getAllSessions()); + } finally { + lock.unlock(); } for (SessionHandle handle : allSessions) { diff --git a/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/session/SessionPoolImplTest.java b/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/session/SessionPoolImplTest.java index 85ebc6937cbe..9e7c5d542ba9 100644 --- a/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/session/SessionPoolImplTest.java +++ b/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/session/SessionPoolImplTest.java @@ -76,6 +76,7 @@ import io.grpc.ServerInterceptor; import io.grpc.Status; import java.io.IOException; +import java.lang.reflect.Field; import java.time.Duration; import java.time.Instant; import java.util.List; @@ -84,10 +85,13 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.LongPredicate; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -679,6 +683,180 @@ public void refreshConfigTest() throws Exception { assertThat(containsHeader).isTrue(); } + private void assertPoolServesVRpc(SessionPoolImpl pool, String label) + throws Exception { + VRpc vrpc = + pool.newCall(FakeDescriptor.SCRIPTED); + UnaryResponseFuture f = new UnaryResponseFuture<>(); + vrpc.start( + SessionFakeScriptedRequest.newBuilder().setTag(0).build(), + VRpcCallContext.create(Deadline.after(15, TimeUnit.SECONDS), true, vrpcTracer), + f); + try { + f.get(15, TimeUnit.SECONDS); + } catch (TimeoutException e) { + throw new AssertionError( + "pool did not serve a vRPC (" + label + ") within 15s — the pool monitor appears wedged", + e); + } + } + + /** + * Reproduces the production thread-dump wedge and verifies the fix. + * + *

The dump (2026-07-23) showed a {@code SessionPoolImpl} whose pool lock had no owner but + * dozens of waiters — an orphaned monitor, the classic signature of a thread that died while + * holding the lock (e.g. a native-thread OOM mid-critical-section). Because the shim probes the + * pool on every RPC via the {@code synchronized hasSession()} / {@code newCall()} path, every + * incoming request then blocked on that lock forever, threads piled up, and the pod shed 100% of + * traffic with no recovery. + * + *

A truly orphaned intrinsic monitor cannot be produced inside a JVM unit test (a dead + * thread's {@code synchronized} block always unwinds), so we simulate the dead owner faithfully: + * a helper thread grabs the pool lock and never lets go. Under the old {@code synchronized + * (this)} design every call below would block on that held monitor indefinitely and this test + * would fail on its timeout. With the {@link java.util.concurrent.locks.ReentrantLock} + bounded + * {@code tryLock} fix, the request-serving hot path degrades to a bounded fast-fail instead of + * hanging, and the pool fully recovers once the lock is released. + */ + @Test + @Timeout(60) + void orphanedPoolLock_hotPathDegradesInsteadOfHangingForever() throws Exception { + sessionPool.start(OpenFakeSessionRequest.getDefaultInstance(), new Metadata()); + // Baseline: a live session exists and the pool serves traffic normally. + assertPoolServesVRpc(sessionPool, "baseline (before lock is orphaned)"); + + ReentrantLock poolLock = extractPoolLock(sessionPool); + + // A separate probe thread stands in for real request threads; the test thread must never touch + // the pool directly, because on the unfixed code those calls would block forever. + ExecutorService probe = Executors.newSingleThreadExecutor(); + // The "dead owner": this thread acquires the pool lock and parks holding it, exactly like the + // thread dump's owner-less-but-waited-on monitor. + ExecutorService deadOwner = Executors.newSingleThreadExecutor(); + CountDownLatch lockHeld = new CountDownLatch(1); + CountDownLatch releaseLock = new CountDownLatch(1); + try { + deadOwner.submit( + () -> { + poolLock.lock(); + try { + lockHeld.countDown(); + releaseLock.await(); // hold the lock until the test releases us + } finally { + poolLock.unlock(); + } + return null; + }); + assertThat(lockHeld.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(poolLock.isLocked()).isTrue(); + assertThat(poolLock.isHeldByCurrentThread()).isFalse(); // owned by the dead-owner thread + + // (1) The per-RPC routing probe must return promptly instead of blocking on the wedged lock. + // On the old synchronized code this get() would time out (the probe thread would be stuck). + Future hasSession = probe.submit(sessionPool::hasSession); + assertThat(hasSession.get(15, TimeUnit.SECONDS)).isFalse(); + + // (2) A brand-new call must also degrade to a bounded fast-fail rather than hang. newCall() + // returns without blocking and start() fast-fails the vRPC with UNAVAILABLE (uncommitted, so + // gax can still retry / fall back to the classic client). + CompletableFuture wedgedResult = new CompletableFuture<>(); + Future starter = + probe.submit( + () -> { + VRpc vrpc = + sessionPool.newCall(FakeDescriptor.SCRIPTED); + vrpc.start( + SessionFakeScriptedRequest.newBuilder().setTag(0).build(), + VRpcCallContext.create(Deadline.after(30, TimeUnit.SECONDS), true, vrpcTracer), + new VRpc.VRpcListener() { + @Override + public void onMessage(SessionFakeScriptedResponse msg) {} + + @Override + public void onClose(VRpcResult result) { + wedgedResult.complete(result); + } + }); + }); + starter.get(15, TimeUnit.SECONDS); // start() itself must return bounded + VRpcResult result = wedgedResult.get(15, TimeUnit.SECONDS); + assertThat(result.getStatus().getCode()).isEqualTo(Status.Code.UNAVAILABLE); + + // (2b) Cancelling a call must also be bounded. PendingVRpc.cancel() takes poolLock only to + // eagerly remove itself from pendingRpcs; on the old code that was an unbounded lock() and + // would park this (callback-path) thread forever on the wedged lock. It must now skip the + // eager removal after the bounded timeout and return promptly. + Future canceller = + probe.submit( + () -> { + VRpc vrpc = + sessionPool.newCall(FakeDescriptor.SCRIPTED); + vrpc.start( + SessionFakeScriptedRequest.newBuilder().setTag(0).build(), + VRpcCallContext.create(Deadline.after(30, TimeUnit.SECONDS), true, vrpcTracer), + new VRpc.VRpcListener() { + @Override + public void onMessage(SessionFakeScriptedResponse msg) {} + + @Override + public void onClose(VRpcResult ignored) {} + }); + vrpc.cancel("cancelled while pool lock is wedged", null); + }); + canceller.get(15, TimeUnit.SECONDS); // cancel() must return bounded, not hang on the lock + } finally { + // Revive the pool: the "dead owner" releases the lock. A healthy pool must resume serving. + releaseLock.countDown(); + deadOwner.shutdown(); + assertThat(deadOwner.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + probe.shutdownNow(); + } + + // (3) Recovery: once the lock is free, the pool serves traffic again — no permanent wedge. + assertPoolServesVRpc(sessionPool, "after the orphaned lock was released"); + } + + /** + * Regression guard for the bounded-lock fix: the timed {@code tryLock(timeout)} used on the hot + * path throws {@code InterruptedException} immediately if the calling thread already has + * its interrupt flag set, even when the lock is free. The old {@code synchronized} never observed + * the interrupt flag, so without the non-interruptible zero-arg {@code tryLock()} fast path a + * pre-existing interrupt (common on reused worker threads) would spuriously shed traffic on a + * perfectly healthy pool. This verifies a healthy pool still serves — and still reports a session + * — when the caller arrives interrupted, and that the interrupt status is preserved. + */ + @Test + @Timeout(60) + void healthyPool_preExistingInterrupt_doesNotSpuriouslyDegrade() throws Exception { + sessionPool.start(OpenFakeSessionRequest.getDefaultInstance(), new Metadata()); + assertPoolServesVRpc(sessionPool, "baseline (before interrupt)"); + + ExecutorService probe = Executors.newSingleThreadExecutor(); + try { + Future interruptStillSet = + probe.submit( + () -> { + Thread.currentThread().interrupt(); // arrive already-interrupted + // A healthy pool must still report its ready session rather than fast-failing. + assertThat(sessionPool.hasSession()).isTrue(); + // The interrupt flag must survive the acquisition (fast path never consumes it). + return Thread.currentThread().isInterrupted(); + }); + assertThat(interruptStillSet.get(15, TimeUnit.SECONDS)).isTrue(); + } finally { + probe.shutdownNow(); + } + + assertPoolServesVRpc(sessionPool, "after an interrupted caller (pool must remain healthy)"); + } + + private static ReentrantLock extractPoolLock(SessionPoolImpl pool) throws Exception { + Field field = SessionPoolImpl.class.getDeclaredField("poolLock"); + field.setAccessible(true); + return (ReentrantLock) field.get(pool); + } + private static class DelayedClientInterceptor implements ClientInterceptor { private final Duration delay; diff --git a/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/session/WatchdogTest.java b/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/session/WatchdogTest.java index c95717ccbb4d..121fae54df22 100644 --- a/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/session/WatchdogTest.java +++ b/java-bigtable/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/internal/session/WatchdogTest.java @@ -60,7 +60,7 @@ void setUp() { sessions = new SessionList(); watchdog = new Watchdog( - new Object(), + new java.util.concurrent.locks.ReentrantLock(), timer, MoreExecutors.directExecutor(), interval,