natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-3.9' into trunk

This commit is contained in:
Stefania Alborghetti 2016-07-29 15:30:57 +08:00
parent cf68431adc 8e775eaaf5
commit bdaa53de4e
8 changed files with 128 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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:

2
bin/cqlsh.py
View File

@ -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,)

41
src/java/org/apache/cassandra/concurrent/StageManager.java
View File

@ -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<Runnable>(1000),
new NamedThreadFactory(Stage.TRACING.getJmxName()),
reh);
return new TracingExecutor(1,
1,
KEEPALIVE,
TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(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<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)
public TracingExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> 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 <T> Future<T> submit(Runnable task, T result)
{
throw new UnsupportedOperationException();
}
@Override
public <T> Future<T> submit(Callable<T> task)
{
throw new UnsupportedOperationException();
}
}
}

10
src/java/org/apache/cassandra/tracing/ExpiredTraceState.java
View File

@ -42,4 +42,14 @@ class ExpiredTraceState extends TraceState
{
delegate.traceImpl(message);
}
protected void waitForPendingEvents()
{
delegate.waitForPendingEvents();
}
TraceState getDelegate()
{
return delegate;
}
}

8
src/java/org/apache/cassandra/tracing/TraceState.java
View File

@ -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();
}
}

59
src/java/org/apache/cassandra/tracing/TraceStateImpl.java
View File

@ -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<Future<?>> 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<Void> 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)

49
src/java/org/apache/cassandra/tracing/TracingImpl.java
View File

@ -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<String, String> 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)
{

3
test/unit/org/apache/cassandra/tracing/TracingTest.java
View File

@ -162,6 +162,9 @@ public final class TracingTest
traces.add(string);
}
protected void waitForPendingEvents()
{
}
};
}