diff --git a/CHANGES.txt b/CHANGES.txt index bdf0db3d5b..27655d2d88 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -37,6 +37,7 @@ Merged from 3.0: * NullPointerExpception when reading/compacting table (CASSANDRA-11988) * Fix problem with undeleteable rows on upgrade to new sstable format (CASSANDRA-12144) Merged from 2.2: + * Wait for tracing events before returning response and query at same consistency level client side (CASSANDRA-11465) * cqlsh copyutil should get host metadata by connected address (CASSANDRA-11979) * Fixed cqlshlib.test.remove_test_db (CASSANDRA-12214) Merged from 2.1: diff --git a/bin/cqlsh.py b/bin/cqlsh.py index 085f5b6468..ee23b29f65 100644 --- a/bin/cqlsh.py +++ b/bin/cqlsh.py @@ -1218,7 +1218,7 @@ class Shell(cmd.Cmd): if self.tracing_enabled: try: - for trace in future.get_all_query_traces(self.max_trace_wait): + for trace in future.get_all_query_traces(max_wait_per=self.max_trace_wait, query_cl=self.consistency_level): print_trace(self, trace) except TraceUnavailable: msg = "Statement trace did not complete within %d seconds; trace data may be incomplete." % (self.session.max_trace_wait,) diff --git a/src/java/org/apache/cassandra/concurrent/StageManager.java b/src/java/org/apache/cassandra/concurrent/StageManager.java index badc527dd2..84b8da6c30 100644 --- a/src/java/org/apache/cassandra/concurrent/StageManager.java +++ b/src/java/org/apache/cassandra/concurrent/StageManager.java @@ -59,7 +59,7 @@ public class StageManager stages.put(Stage.TRACING, tracingExecutor()); } - private static ExecuteOnlyExecutor tracingExecutor() + private static LocalAwareExecutorService tracingExecutor() { RejectedExecutionHandler reh = new RejectedExecutionHandler() { @@ -68,13 +68,13 @@ public class StageManager MessagingService.instance().incrementDroppedMessages(MessagingService.Verb._TRACE); } }; - return new ExecuteOnlyExecutor(1, - 1, - KEEPALIVE, - TimeUnit.SECONDS, - new ArrayBlockingQueue(1000), - new NamedThreadFactory(Stage.TRACING.getJmxName()), - reh); + return new TracingExecutor(1, + 1, + KEEPALIVE, + TimeUnit.SECONDS, + new ArrayBlockingQueue(1000), + new NamedThreadFactory(Stage.TRACING.getJmxName()), + reh); } private static JMXEnabledThreadPoolExecutor multiThreadedStage(Stage stage, int numThreads) @@ -113,12 +113,11 @@ public class StageManager } /** - * A TPE that disallows submit so that we don't need to worry about unwrapping exceptions on the - * tracing stage. See CASSANDRA-1123 for background. + * The executor used for tracing. */ - private static class ExecuteOnlyExecutor extends ThreadPoolExecutor implements LocalAwareExecutorService + private static class TracingExecutor extends ThreadPoolExecutor implements LocalAwareExecutorService { - public ExecuteOnlyExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) + public TracingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler) { super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler); } @@ -133,23 +132,5 @@ public class StageManager { execute(command); } - - @Override - public Future submit(Runnable task) - { - throw new UnsupportedOperationException(); - } - - @Override - public Future submit(Runnable task, T result) - { - throw new UnsupportedOperationException(); - } - - @Override - public Future submit(Callable task) - { - throw new UnsupportedOperationException(); - } } } diff --git a/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java b/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java index fbe2c331bf..9230d38f67 100644 --- a/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java +++ b/src/java/org/apache/cassandra/tracing/ExpiredTraceState.java @@ -42,4 +42,14 @@ class ExpiredTraceState extends TraceState { delegate.traceImpl(message); } + + protected void waitForPendingEvents() + { + delegate.waitForPendingEvents(); + } + + TraceState getDelegate() + { + return delegate; + } } diff --git a/src/java/org/apache/cassandra/tracing/TraceState.java b/src/java/org/apache/cassandra/tracing/TraceState.java index 5365d0964d..b4eff6b386 100644 --- a/src/java/org/apache/cassandra/tracing/TraceState.java +++ b/src/java/org/apache/cassandra/tracing/TraceState.java @@ -111,6 +111,8 @@ public abstract class TraceState implements ProgressEventNotifier public synchronized void stop() { + waitForPendingEvents(); + status = Status.STOPPED; notifyAll(); } @@ -179,6 +181,11 @@ public abstract class TraceState implements ProgressEventNotifier protected abstract void traceImpl(String message); + protected void waitForPendingEvents() + { + // if tracing events are asynchronous, then you can use this method to wait for them to complete + } + public boolean acquireReference() { while (true) @@ -193,6 +200,7 @@ public abstract class TraceState implements ProgressEventNotifier public int releaseReference() { + waitForPendingEvents(); return references.decrementAndGet(); } } diff --git a/src/java/org/apache/cassandra/tracing/TraceStateImpl.java b/src/java/org/apache/cassandra/tracing/TraceStateImpl.java index 113ebb7a10..55e8389c8c 100644 --- a/src/java/org/apache/cassandra/tracing/TraceStateImpl.java +++ b/src/java/org/apache/cassandra/tracing/TraceStateImpl.java @@ -19,7 +19,16 @@ package org.apache.cassandra.tracing; import java.net.InetAddress; import java.util.Collections; +import java.util.Set; import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.concurrent.StageManager; @@ -27,6 +36,7 @@ import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.Mutation; import org.apache.cassandra.exceptions.OverloadedException; import org.apache.cassandra.service.StorageProxy; +import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.WrappedRunnable; /** @@ -35,6 +45,12 @@ import org.apache.cassandra.utils.WrappedRunnable; */ public class TraceStateImpl extends TraceState { + private static final Logger logger = LoggerFactory.getLogger(TraceStateImpl.class); + private static final int WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS = + Integer.valueOf(System.getProperty("cassandra.wait_for_tracing_events_timeout_secs", "1")); + + private final Set> pendingFutures = ConcurrentHashMap.newKeySet(); + public TraceStateImpl(InetAddress coordinator, UUID sessionId, Tracing.TraceType traceType) { super(coordinator, sessionId, traceType); @@ -46,17 +62,54 @@ public class TraceStateImpl extends TraceState final int elapsed = elapsed(); executeMutation(TraceKeyspace.makeEventMutation(sessionIdBytes, message, elapsed, threadName, ttl)); + if (logger.isTraceEnabled()) + logger.trace("Adding <{}> to trace events", message); } - static void executeMutation(final Mutation mutation) + /** + * Wait on submitted futures + */ + protected void waitForPendingEvents() { - StageManager.getStage(Stage.TRACING).execute(new WrappedRunnable() + if (WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS <= 0) + return; + + try + { + if (logger.isTraceEnabled()) + logger.trace("Waiting for up to {} seconds for {} trace events to complete", + +WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, pendingFutures.size()); + + CompletableFuture.allOf(pendingFutures.toArray(new CompletableFuture[pendingFutures.size()])) + .get(WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS, TimeUnit.SECONDS); + } + catch (TimeoutException ex) + { + if (logger.isTraceEnabled()) + logger.trace("Failed to wait for tracing events to complete in {} seconds", + WAIT_FOR_PENDING_EVENTS_TIMEOUT_SECS); + } + catch (Throwable t) + { + JVMStabilityInspector.inspectThrowable(t); + logger.error("Got exception whilst waiting for tracing events to complete", t); + } + } + + + void executeMutation(final Mutation mutation) + { + CompletableFuture fut = CompletableFuture.runAsync(new WrappedRunnable() { protected void runMayThrow() { mutateWithCatch(mutation); } - }); + }, StageManager.getStage(Stage.TRACING)); + + boolean ret = pendingFutures.add(fut); + if (!ret) + logger.warn("Failed to insert pending future, tracing synchronization may not work"); } static void mutateWithCatch(Mutation mutation) diff --git a/src/java/org/apache/cassandra/tracing/TracingImpl.java b/src/java/org/apache/cassandra/tracing/TracingImpl.java index 52ac183cf6..4f69584017 100644 --- a/src/java/org/apache/cassandra/tracing/TracingImpl.java +++ b/src/java/org/apache/cassandra/tracing/TracingImpl.java @@ -28,10 +28,6 @@ import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.concurrent.StageManager; import org.apache.cassandra.utils.WrappedRunnable; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - - /** * A trace session context. Able to track and store trace sessions. A session is usually a user initiated query, and may @@ -39,31 +35,62 @@ import org.slf4j.LoggerFactory; */ class TracingImpl extends Tracing { - private static final Logger logger = LoggerFactory.getLogger(TracingImpl.class); - public void stopSessionImpl() { - TraceState state = get(); + final TraceStateImpl state = getStateImpl(); + if (state == null) + return; + int elapsed = state.elapsed(); ByteBuffer sessionId = state.sessionIdBytes; int ttl = state.ttl; - TraceStateImpl.executeMutation(TraceKeyspace.makeStopSessionMutation(sessionId, elapsed, ttl)); + + state.executeMutation(TraceKeyspace.makeStopSessionMutation(sessionId, elapsed, ttl)); } public TraceState begin(final String request, final InetAddress client, final Map parameters) { assert isTracing(); - final TraceState state = get(); + final TraceStateImpl state = getStateImpl(); + assert state != null; + final long startedAt = System.currentTimeMillis(); final ByteBuffer sessionId = state.sessionIdBytes; final String command = state.traceType.toString(); final int ttl = state.ttl; - TraceStateImpl.executeMutation(TraceKeyspace.makeStartSessionMutation(sessionId, client, parameters, request, startedAt, command, ttl)); - + state.executeMutation(TraceKeyspace.makeStartSessionMutation(sessionId, client, parameters, request, startedAt, command, ttl)); return state; } + /** + * Convert the abstract tracing state to its implementation. + * + * Expired states are not put in the sessions but the check is for extra safety. + * + * @return the state converted to its implementation, or null + */ + private TraceStateImpl getStateImpl() + { + TraceState state = get(); + if (state == null) + return null; + + if (state instanceof ExpiredTraceState) + { + ExpiredTraceState expiredTraceState = (ExpiredTraceState) state; + state = expiredTraceState.getDelegate(); + } + + if (state instanceof TraceStateImpl) + { + return (TraceStateImpl)state; + } + + assert false : "TracingImpl states should be of type TraceStateImpl"; + return null; + } + @Override protected TraceState newTraceState(InetAddress coordinator, UUID sessionId, TraceType traceType) { diff --git a/test/unit/org/apache/cassandra/tracing/TracingTest.java b/test/unit/org/apache/cassandra/tracing/TracingTest.java index 1b0e507270..30521c0bf0 100644 --- a/test/unit/org/apache/cassandra/tracing/TracingTest.java +++ b/test/unit/org/apache/cassandra/tracing/TracingTest.java @@ -162,6 +162,9 @@ public final class TracingTest traces.add(string); } + protected void waitForPendingEvents() + { + } }; }