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

[CASSANDRA-16924] CEP-10 Phase 1: Mockable Blocking Concurrency Primitives 

patch by Benedict; reviewed by Sam Tunnicliffe and Aleksei Zotov for CASSANDRA-16924

Co-authored-by: Benedict Elliott Smith <benedict@apache.org>
Co-authored-by: Sam Tunnicliffe <samt@apache.org>
This commit is contained in:
Benedict Elliott Smith 2021-07-28 19:24:43 +01:00 committed by Benedict Elliott Smith
parent 2e2db4dc40
commit e5b92e1088
304 changed files with 3665 additions and 1993 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
checkstyle_suppressions.xml
View File

@ -22,5 +22,5 @@
"https://checkstyle.org/dtds/suppressions_1_1.dtd">
<suppressions>
<suppress checks="RegexpSinglelineJava" files="Clock\.java"/>
<suppress checks="RegexpSinglelineJava" files="Clock\.java|Semaphore\.java"/>
</suppressions>

30
src/java/org/apache/cassandra/cache/AutoSavingCache.java vendored
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.cache;
import java.io.*;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
@ -46,7 +45,6 @@ import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.db.compaction.CompactionInfo.Unit;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.io.util.CorruptFileException;
import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.utils.JVMStabilityInspector;
@ -162,26 +160,14 @@ public class AutoSavingCache<K extends CacheKey, V> extends InstrumentingCache<K
final ListeningExecutorService es = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
final long start = nanoTime();
ListenableFuture<Integer> cacheLoad = es.submit(new Callable<Integer>()
{
@Override
public Integer call()
{
return loadSaved();
}
});
cacheLoad.addListener(new Runnable()
{
@Override
public void run()
{
if (size() > 0)
logger.info("Completed loading ({} ms; {} keys) {} cache",
TimeUnit.NANOSECONDS.toMillis(nanoTime() - start),
CacheService.instance.keyCache.size(),
cacheType);
es.shutdown();
}
ListenableFuture<Integer> cacheLoad = es.submit(this::loadSaved);
cacheLoad.addListener(() -> {
if (size() > 0)
logger.info("Completed loading ({} ms; {} keys) {} cache",
TimeUnit.NANOSECONDS.toMillis(nanoTime() - start),
CacheService.instance.keyCache.size(),
cacheType);
es.shutdown();
}, MoreExecutors.directExecutor());
return cacheLoad;

7
src/java/org/apache/cassandra/concurrent/AbstractLocalAwareExecutorService.java
View File

@ -26,10 +26,11 @@ import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.cassandra.utils.concurrent.Condition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import org.apache.cassandra.utils.JVMStabilityInspector;
import static org.apache.cassandra.tracing.Tracing.isTracing;
@ -140,7 +141,7 @@ public abstract class AbstractLocalAwareExecutorService implements LocalAwareExe
}
}
class FutureTask<T> extends SimpleCondition implements Future<T>, Runnable
class FutureTask<T> extends Condition.Async implements Future<T>, Runnable
{
private boolean failure;
private Object result = this;
@ -187,7 +188,7 @@ public abstract class AbstractLocalAwareExecutorService implements LocalAwareExe
public boolean isDone()
{
return isSignaled();
return isSignalled();
}
public T get() throws InterruptedException, ExecutionException

7
src/java/org/apache/cassandra/concurrent/JMXEnabledSingleThreadExecutor.java
View File

@ -18,15 +18,16 @@
package org.apache.cassandra.concurrent;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
public class JMXEnabledSingleThreadExecutor extends JMXEnabledThreadPoolExecutor
{
public JMXEnabledSingleThreadExecutor(String threadPoolName, String jmxPath)
{
super(1, Integer.MAX_VALUE, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new SingleThreadFactory(threadPoolName), jmxPath);
super(1, Integer.MAX_VALUE, SECONDS, newBlockingQueue(), new SingleThreadFactory(threadPoolName), jmxPath);
}
@Override

2
src/java/org/apache/cassandra/concurrent/NamedThreadFactory.java
View File

@ -22,9 +22,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.annotations.VisibleForTesting;
import io.netty.util.concurrent.FastThreadLocal;
import io.netty.util.concurrent.FastThreadLocalThread;
import org.apache.cassandra.utils.memory.BufferPool;
/**
* This class is an implementation of the <i>ThreadFactory</i> interface. This

18
src/java/org/apache/cassandra/concurrent/SEPExecutor.java
View File

@ -26,14 +26,16 @@ import java.util.concurrent.atomic.AtomicLong;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.concurrent.Condition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.metrics.ThreadPoolMetrics;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import static org.apache.cassandra.concurrent.SEPExecutor.TakeTaskPermitResult.*;
import static org.apache.cassandra.concurrent.SEPWorker.Work;
import static org.apache.cassandra.utils.concurrent.Condition.newOneTimeCondition;
public class SEPExecutor extends AbstractLocalAwareExecutorService implements SEPExecutorMBean
{
@ -55,7 +57,7 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService implements SE
private final AtomicLong completedTasks = new AtomicLong();
volatile boolean shuttingDown = false;
final SimpleCondition shutdown = new SimpleCondition();
final Condition shutdown = newOneTimeCondition();
// TODO: see if other queue implementations might improve throughput
protected final ConcurrentLinkedQueue<FutureTask<?>> tasks = new ConcurrentLinkedQueue<>();
@ -144,14 +146,14 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService implements SE
// Work permits are negative when the pool is reducing in size. Atomically
// adjust the number of work permits so there is no race of multiple SEPWorkers
// exiting. On conflicting update, recheck.
result = TakeTaskPermitResult.RETURNED_WORK_PERMIT;
result = RETURNED_WORK_PERMIT;
updated = updateWorkPermits(current, workPermits + 1);
}
else
{
if (taskPermits == 0)
return TakeTaskPermitResult.NONE_AVAILABLE;
result = TakeTaskPermitResult.TOOK_PERMIT;
return NONE_AVAILABLE;
result = TOOK_PERMIT;
updated = updateTaskPermits(current, taskPermits - 1);
}
if (permits.compareAndSet(current, updated))
@ -234,7 +236,7 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService implements SE
{
shutdown();
List<Runnable> aborted = new ArrayList<>();
while (takeTaskPermit(false) == TakeTaskPermitResult.TOOK_PERMIT)
while (takeTaskPermit(false) == TOOK_PERMIT)
aborted.add(tasks.poll());
return aborted;
}
@ -246,7 +248,7 @@ public class SEPExecutor extends AbstractLocalAwareExecutorService implements SE
public boolean isTerminated()
{
return shuttingDown && shutdown.isSignaled();
return shuttingDown && shutdown.isSignalled();
}
public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException

7
src/java/org/apache/cassandra/concurrent/Stage.java
View File

@ -25,7 +25,6 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
@ -45,7 +44,9 @@ import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.FBUtilities;
import static java.util.concurrent.TimeUnit.SECONDS;
import static java.util.stream.Collectors.toMap;
import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
public enum Stage
{
@ -182,8 +183,8 @@ public enum Stage
{
return new JMXEnabledThreadPoolExecutor(numThreads,
KEEP_ALIVE_SECONDS,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(),
SECONDS,
newBlockingQueue(),
new NamedThreadFactory(jmxName),
jmxType);
}

1
src/java/org/apache/cassandra/cql3/CQL3Type.java
View File

@ -24,7 +24,6 @@ import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.db.marshal.CollectionType.Kind;

1
src/java/org/apache/cassandra/cql3/functions/AbstractFunction.java
View File

@ -27,7 +27,6 @@ import org.apache.cassandra.cql3.CQL3Type;
import org.apache.cassandra.cql3.CQL3Type.Tuple;
import org.apache.cassandra.cql3.ColumnSpecification;
import org.apache.cassandra.cql3.CqlBuilder;
import org.apache.cassandra.cql3.CqlBuilder.Appender;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.commons.lang3.text.StrBuilder;

14
src/java/org/apache/cassandra/cql3/functions/UDFExecutorService.java
View File

@ -17,12 +17,12 @@
*/
package org.apache.cassandra.cql3.functions;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.utils.FBUtilities;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.cassandra.utils.FBUtilities.getAvailableProcessors;
import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
/**
* Executor service which exposes stats via JMX, but which doesn't reference
@ -35,10 +35,10 @@ final class UDFExecutorService extends JMXEnabledThreadPoolExecutor
UDFExecutorService(NamedThreadFactory threadFactory, String jmxPath)
{
super(FBUtilities.getAvailableProcessors(),
super(getAvailableProcessors(),
KEEPALIVE,
TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(),
MILLISECONDS,
newBlockingQueue(),
threadFactory,
jmxPath);
}

5
src/java/org/apache/cassandra/cql3/functions/UDFunction.java
View File

@ -58,6 +58,7 @@ import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static com.google.common.collect.Iterables.any;
import static com.google.common.collect.Iterables.transform;
@ -512,7 +513,7 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
catch (InterruptedException e)
{
Thread.currentThread().interrupt();
throw new RuntimeException(e);
throw new UncheckedInterruptedException(e);
}
catch (ExecutionException e)
{
@ -538,7 +539,7 @@ public abstract class UDFunction extends AbstractFunction implements ScalarFunct
catch (InterruptedException e1)
{
Thread.currentThread().interrupt();
throw new RuntimeException(e);
throw new UncheckedInterruptedException(e1);
}
catch (ExecutionException e1)
{

2
src/java/org/apache/cassandra/cql3/functions/types/TypeCodec.java
View File

@ -36,8 +36,6 @@ import com.google.common.io.ByteStreams;
import com.google.common.reflect.TypeToken;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.cql3.functions.types.DataType.CollectionType;
import org.apache.cassandra.cql3.functions.types.DataType.Name;
import org.apache.cassandra.cql3.functions.types.exceptions.InvalidTypeException;
import org.apache.cassandra.cql3.functions.types.utils.Bytes;
import org.apache.cassandra.utils.vint.VIntCoding;

2
src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
View File

@ -21,9 +21,7 @@ import java.nio.ByteBuffer;
import java.util.*;
import com.google.common.collect.HashMultiset;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

2
src/java/org/apache/cassandra/cql3/statements/schema/CreateFunctionStatement.java
View File

@ -26,8 +26,6 @@ import com.google.common.collect.Lists;
import org.apache.cassandra.audit.AuditLogContext;
import org.apache.cassandra.audit.AuditLogEntryType;
import org.apache.cassandra.auth.FunctionResource;
import org.apache.cassandra.auth.IResource;
import org.apache.cassandra.auth.*;
import org.apache.cassandra.cql3.CQL3Type;
import org.apache.cassandra.cql3.CQLStatement;

1
src/java/org/apache/cassandra/cql3/statements/schema/DropIndexStatement.java
View File

@ -22,7 +22,6 @@ import org.apache.cassandra.audit.AuditLogEntryType;
import org.apache.cassandra.auth.Permission;
import org.apache.cassandra.cql3.CQLStatement;
import org.apache.cassandra.cql3.QualifiedName;
import org.apache.cassandra.schema.Diff;
import org.apache.cassandra.schema.*;
import org.apache.cassandra.schema.KeyspaceMetadata.KeyspaceDiff;
import org.apache.cassandra.schema.Keyspaces.KeyspacesDiff;

1
src/java/org/apache/cassandra/db/BufferClustering.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.db;
import java.nio.ByteBuffer;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
/**

1
src/java/org/apache/cassandra/db/BufferClusteringBound.java
View File

@ -22,7 +22,6 @@ import java.nio.ByteBuffer;
import com.google.common.base.Preconditions;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;

1
src/java/org/apache/cassandra/db/BufferClusteringBoundary.java
View File

@ -22,7 +22,6 @@ import java.nio.ByteBuffer;
import com.google.common.base.Preconditions;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;

14
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -92,14 +92,18 @@ import org.apache.cassandra.streaming.TableStreamManager;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.concurrent.Refs;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import org.apache.cassandra.utils.memory.MemtableAllocator;
import static com.google.common.base.Throwables.propagate;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.db.commitlog.CommitLogPosition.NONE;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.Throwables.maybeFail;
import static org.apache.cassandra.utils.Throwables.merge;
import static org.apache.cassandra.utils.concurrent.CountDownLatch.newCountDownLatch;
public class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
@ -953,7 +957,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
*/
private final class PostFlush implements Callable<CommitLogPosition>
{
final CountDownLatch latch = new CountDownLatch(1);
final org.apache.cassandra.utils.concurrent.CountDownLatch latch = newCountDownLatch(1);
final List<Memtable> memtables;
volatile Throwable flushFailure = null;
@ -972,10 +976,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
catch (InterruptedException e)
{
throw new IllegalStateException();
throw new UncheckedInterruptedException(e);
}
CommitLogPosition commitLogUpperBound = CommitLogPosition.NONE;
CommitLogPosition commitLogUpperBound = NONE;
// If a flush errored out but the error was ignored, make sure we don't discard the commit log.
if (flushFailure == null && !memtables.isEmpty())
{
@ -987,7 +991,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
metric.pendingFlushes.dec();
if (flushFailure != null)
throw Throwables.propagate(flushFailure);
throw propagate(flushFailure);
return commitLogUpperBound;
}
@ -1091,7 +1095,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
logger.trace("Flush task {}@{} signaling post flush task", hashCode(), name);
// signal the post-flush we've done our work
postFlush.latch.countDown();
postFlush.latch.decrement();
if (logger.isTraceEnabled())
logger.trace("Flush task task {}@{} finished", hashCode(), name);

1
src/java/org/apache/cassandra/db/DiskBoundaryManager.java
View File

@ -21,7 +21,6 @@ package org.apache.cassandra.db;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

3
src/java/org/apache/cassandra/db/Keyspace.java
View File

@ -72,6 +72,7 @@ import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
@ -613,7 +614,7 @@ public class Keyspace
}
catch (InterruptedException e)
{
// Just continue
throw new UncheckedInterruptedException(e);
}
continue;
}

1
src/java/org/apache/cassandra/db/MutableDeletionInfo.java
View File

@ -24,7 +24,6 @@ import com.google.common.base.Objects;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.rows.EncodingStats;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.memory.AbstractAllocator;

1
src/java/org/apache/cassandra/db/Slice.java
View File

@ -22,7 +22,6 @@ import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.ByteArrayAccessor;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;

2
src/java/org/apache/cassandra/db/SnapshotCommand.java
View File

@ -22,8 +22,6 @@ import java.io.IOException;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.Verb;
public class SnapshotCommand
{

1
src/java/org/apache/cassandra/db/SystemKeyspace.java
View File

@ -59,7 +59,6 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.RestorableMeter;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.*;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.paxos.Commit;
import org.apache.cassandra.service.paxos.PaxosState;

1
src/java/org/apache/cassandra/db/UnfilteredDeserializer.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.db;
import java.io.IOException;
import org.apache.cassandra.db.marshal.ByteArrayAccessor;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.io.util.DataInputPlus;

1
src/java/org/apache/cassandra/db/aggregation/GroupingState.java
View File

@ -23,7 +23,6 @@ import java.nio.ByteBuffer;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ClusteringComparator;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.marshal.ByteArrayAccessor;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.ByteBufferUtil;

11
src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogSegmentManager.java
View File

@ -24,6 +24,7 @@ import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BooleanSupplier;
import com.codahale.metrics.Timer.Context;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.util.concurrent.*;
import org.slf4j.Logger;
@ -40,9 +41,11 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static org.apache.cassandra.db.commitlog.CommitLogSegment.Allocation;
import static org.apache.cassandra.utils.concurrent.WaitQueue.newWaitQueue;
/**
* Performs eager-creation of commit log segments in a background thread. All the
@ -62,7 +65,7 @@ public abstract class AbstractCommitLogSegmentManager
*/
private volatile CommitLogSegment availableSegment = null;
private final WaitQueue segmentPrepared = new WaitQueue();
private final WaitQueue segmentPrepared = newWaitQueue();
/** Active segments, containing unflushed data. The tail of this queue is the one we allocate writes to */
private final ConcurrentLinkedQueue<CommitLogSegment> activeSegments = new ConcurrentLinkedQueue<>();
@ -89,7 +92,7 @@ public abstract class AbstractCommitLogSegmentManager
protected final CommitLog commitLog;
private volatile boolean shutdown;
private final BooleanSupplier managerThreadWaitCondition = () -> (availableSegment == null && !atSegmentBufferLimit()) || shutdown;
private final WaitQueue managerThreadWaitQueue = new WaitQueue();
private final WaitQueue managerThreadWaitQueue = newWaitQueue();
private volatile SimpleCachedBufferPool bufferPool;
@ -265,7 +268,7 @@ public abstract class AbstractCommitLogSegmentManager
{
do
{
WaitQueue.Signal prepared = segmentPrepared.register(commitLog.metrics.waitingOnSegmentAllocation.time());
WaitQueue.Signal prepared = segmentPrepared.register(commitLog.metrics.waitingOnSegmentAllocation.time(), Context::stop);
if (availableSegment == null && allocatingFrom == currentAllocatingFrom)
prepared.awaitUninterruptibly();
else
@ -430,7 +433,7 @@ public abstract class AbstractCommitLogSegmentManager
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
throw new UncheckedInterruptedException(e);
}
for (CommitLogSegment segment : activeSegments)

7
src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
View File

@ -25,7 +25,6 @@ import com.google.common.annotations.VisibleForTesting;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.codahale.metrics.Timer.Context;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.db.commitlog.CommitLogSegment.Allocation;
@ -33,8 +32,10 @@ import org.apache.cassandra.utils.MonotonicClock;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static com.codahale.metrics.Timer.*;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.concurrent.WaitQueue.newWaitQueue;
public abstract class AbstractCommitLogService
{
@ -55,7 +56,7 @@ public abstract class AbstractCommitLogService
protected final AtomicLong pending = new AtomicLong(0);
// signal that writers can wait on to be notified of a completed sync
protected final WaitQueue syncComplete = new WaitQueue();
protected final WaitQueue syncComplete = newWaitQueue();
final CommitLog commitLog;
private final String name;
@ -304,7 +305,7 @@ public abstract class AbstractCommitLogService
{
do
{
WaitQueue.Signal signal = context != null ? syncComplete.register(context) : syncComplete.register();
WaitQueue.Signal signal = context != null ? syncComplete.register(context, Context::stop) : syncComplete.register();
if (lastSyncedAt < syncTime)
signal.awaitUninterruptibly();
else

5
src/java/org/apache/cassandra/db/commitlog/CommitLog.java
View File

@ -46,6 +46,7 @@ import org.apache.cassandra.security.EncryptionContext;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static org.apache.cassandra.db.commitlog.CommitLogSegment.Allocation;
import static org.apache.cassandra.db.commitlog.CommitLogSegment.CommitLogSegmentFileComparator;
@ -469,7 +470,7 @@ public class CommitLog implements CommitLogMBean
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
throw new UncheckedInterruptedException(e);
}
segmentManager.stopUnsafe(deleteSegments);
CommitLogSegment.resetReplayLimit();
@ -549,7 +550,7 @@ public class CommitLog implements CommitLogMBean
*/
public boolean useEncryption()
{
return encryptionContext.isEnabled();
return encryptionContext != null && encryptionContext.isEnabled();
}
/**

4
src/java/org/apache/cassandra/db/commitlog/CommitLogArchiver.java
View File

@ -38,6 +38,8 @@ import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.schema.CompressionParams;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.WrappedRunnable;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -193,7 +195,7 @@ public class CommitLogArchiver
}
catch (InterruptedException e)
{
throw new AssertionError(e);
throw new UncheckedInterruptedException(e);
}
catch (ExecutionException e)
{

5
src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java
View File

@ -48,8 +48,10 @@ import org.apache.cassandra.utils.IntegerInterval;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static com.codahale.metrics.Timer.*;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.FBUtilities.updateChecksumInt;
import static org.apache.cassandra.utils.concurrent.WaitQueue.newWaitQueue;
/*
* A single commit log file on disk. Manages creation of the file and writing mutations to disk,
@ -111,7 +113,7 @@ public abstract class CommitLogSegment
private int endOfBuffer;
// a signal for writers to wait on to confirm the log message they provided has been written to disk
private final WaitQueue syncComplete = new WaitQueue();
private final WaitQueue syncComplete = newWaitQueue();
// a map of Cf->dirty interval in this segment; if interval is not covered by the clean set, the log contains unflushed data
private final NonBlockingHashMap<TableId, IntegerInterval> tableDirty = new NonBlockingHashMap<>(1024);
@ -511,7 +513,6 @@ public abstract class CommitLogSegment
while (lastSyncedOffset < position)
{
WaitQueue.Signal signal = syncComplete.register();
if (lastSyncedOffset < position)
signal.awaitUninterruptibly();
else

2
src/java/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerStandard.java
View File

@ -18,8 +18,6 @@
package org.apache.cassandra.db.commitlog;
import java.io.File;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.io.util.FileUtils;

1
src/java/org/apache/cassandra/db/commitlog/EncryptedSegment.java
View File

@ -27,7 +27,6 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.compress.ICompressor;
import org.apache.cassandra.security.EncryptionUtils;
import org.apache.cassandra.security.EncryptionContext;

1
src/java/org/apache/cassandra/db/compaction/AbstractStrategyHolder.java
View File

@ -31,7 +31,6 @@ import com.google.common.base.Preconditions;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.index.Index;

1
src/java/org/apache/cassandra/db/compaction/CompactionManager.java
View File

@ -80,7 +80,6 @@ import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.PreviewKind;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.concurrent.Refs;
import static java.util.Collections.singleton;

1
src/java/org/apache/cassandra/db/compaction/CompactionStrategyHolder.java
View File

@ -29,7 +29,6 @@ import com.google.common.collect.Iterables;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.index.Index;

1
src/java/org/apache/cassandra/db/compaction/PendingRepairHolder.java
View File

@ -30,7 +30,6 @@ import com.google.common.collect.Iterables;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.index.Index;

1
src/java/org/apache/cassandra/db/compaction/SSTableSplitter.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.db.compaction;
import java.util.*;
import java.util.function.LongPredicate;
import java.util.function.Predicate;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.compaction.writers.CompactionAwareWriter;

1
src/java/org/apache/cassandra/db/compaction/Upgrader.java
View File

@ -20,7 +20,6 @@ package org.apache.cassandra.db.compaction;
import java.io.File;
import java.util.*;
import java.util.function.LongPredicate;
import java.util.function.Predicate;
import com.google.common.base.Throwables;
import com.google.common.collect.Sets;

1
src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.db.compaction.writers;
import java.io.File;
import java.util.Collection;
import java.util.List;
import java.util.Set;

1
src/java/org/apache/cassandra/db/lifecycle/ILifecycleTransaction.java
View File

@ -21,7 +21,6 @@ package org.apache.cassandra.db.lifecycle;
import java.util.Collection;
import java.util.Set;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.utils.concurrent.Transactional;

1
src/java/org/apache/cassandra/db/marshal/DurationType.java
View File

@ -26,7 +26,6 @@ import org.apache.cassandra.cql3.Term;
import org.apache.cassandra.serializers.DurationSerializer;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.serializers.TypeSerializer;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.ByteBufferUtil;
/**

1
src/java/org/apache/cassandra/db/marshal/SimpleDateType.java
View File

@ -23,7 +23,6 @@ import org.apache.cassandra.cql3.CQL3Type;
import org.apache.cassandra.cql3.Constants;
import org.apache.cassandra.cql3.Duration;
import org.apache.cassandra.cql3.Term;
import org.apache.cassandra.cql3.statements.RequestValidations;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.serializers.SimpleDateSerializer;
import org.apache.cassandra.serializers.TypeSerializer;

1
src/java/org/apache/cassandra/db/marshal/TimeType.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.db.marshal;
import java.nio.ByteBuffer;
import java.time.LocalTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import org.apache.cassandra.cql3.Constants;

1
src/java/org/apache/cassandra/db/marshal/TimeUUIDType.java
View File

@ -21,7 +21,6 @@ import java.nio.ByteBuffer;
import java.util.UUID;
import org.apache.cassandra.cql3.CQL3Type;
import org.apache.cassandra.cql3.ColumnSpecification;
import org.apache.cassandra.cql3.Constants;
import org.apache.cassandra.cql3.Term;
import org.apache.cassandra.serializers.TypeSerializer;

1
src/java/org/apache/cassandra/db/marshal/TimestampType.java
View File

@ -23,7 +23,6 @@ import java.util.Date;
import org.apache.cassandra.cql3.Constants;
import org.apache.cassandra.cql3.Duration;
import org.apache.cassandra.cql3.Term;
import org.apache.cassandra.cql3.statements.RequestValidations;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

4
src/java/org/apache/cassandra/db/monitoring/MonitoringTask.java
View File

@ -25,7 +25,6 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@ -41,6 +40,7 @@ import org.apache.cassandra.utils.NoSpamLogger;
import static java.lang.System.getProperty;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.utils.MonotonicClock.approxTime;
import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
/**
* A task for monitoring in progress operations, currently only read queries, and aborting them if they time out.
@ -207,7 +207,7 @@ class MonitoringTask
OperationsQueue(int maxOperations)
{
this.maxOperations = maxOperations;
this.queue = maxOperations > 0 ? new ArrayBlockingQueue<>(maxOperations) : new LinkedBlockingQueue<>();
this.queue = maxOperations > 0 ? newBlockingQueue(maxOperations) : newBlockingQueue();
this.numDroppedOperations = new AtomicLong();
}

1
src/java/org/apache/cassandra/db/partitions/Partition.java
View File

@ -22,7 +22,6 @@ import java.util.NavigableSet;
import javax.annotation.Nullable;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.Slices;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.db.filter.ColumnFilter;

1
src/java/org/apache/cassandra/db/partitions/PurgeFunction.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.db.partitions;
import java.util.function.LongPredicate;
import java.util.function.Predicate;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.*;

2
src/java/org/apache/cassandra/db/rows/AbstractRangeTombstoneMarker.java
View File

@ -17,8 +17,8 @@
*/
package org.apache.cassandra.db.rows;
import org.apache.cassandra.db.ClusteringBoundOrBoundary;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.ClusteringBoundOrBoundary;
public abstract class AbstractRangeTombstoneMarker<B extends ClusteringBoundOrBoundary<?>> implements RangeTombstoneMarker
{

1
src/java/org/apache/cassandra/db/rows/ThrottledUnfilteredIterator.java
View File

@ -17,7 +17,6 @@
*/
package org.apache.cassandra.db.rows;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;

1
src/java/org/apache/cassandra/db/streaming/CassandraCompressedStreamReader.java
View File

@ -24,7 +24,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.io.compress.CompressionMetadata;
import org.apache.cassandra.io.sstable.SSTableMultiWriter;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputPlus;

2
src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java
View File

@ -28,7 +28,7 @@ import com.google.common.collect.Iterables;
import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.streaming.StreamReceiveTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

1
src/java/org/apache/cassandra/db/view/ViewBuilderTask.java
View File

@ -29,7 +29,6 @@ import java.util.concurrent.atomic.AtomicLong;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Objects;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterators;
import com.google.common.collect.PeekingIterator;
import com.google.common.util.concurrent.Futures;

1
src/java/org/apache/cassandra/db/view/ViewUtils.java
View File

@ -29,7 +29,6 @@ import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.EndpointsForToken;
import org.apache.cassandra.locator.NetworkTopologyStrategy;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.utils.FBUtilities;
public final class ViewUtils
{

1
src/java/org/apache/cassandra/dht/ByteOrderedPartitioner.java
View File

@ -37,7 +37,6 @@ import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

1
src/java/org/apache/cassandra/dht/Datacenters.java
View File

@ -25,7 +25,6 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.locator.IEndpointSnitch;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.FBUtilities;
public class Datacenters
{

1
src/java/org/apache/cassandra/dht/StreamStateStore.java
View File

@ -30,7 +30,6 @@ import org.apache.cassandra.streaming.StreamEvent;
import org.apache.cassandra.streaming.StreamEventHandler;
import org.apache.cassandra.streaming.StreamRequest;
import org.apache.cassandra.streaming.StreamState;
import org.apache.cassandra.utils.Pair;
/**
* Store and update available ranges (data already received) to system keyspace.

3
src/java/org/apache/cassandra/diag/DiagnosticEventService.java
View File

@ -18,7 +18,6 @@
package org.apache.cassandra.diag;
import java.io.Serializable;
import java.lang.management.ManagementFactory;
import java.util.Collection;
import java.util.Map;
import java.util.Objects;
@ -27,8 +26,6 @@ import java.util.SortedMap;
import java.util.function.Consumer;
import javax.annotation.Nullable;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.collect.ImmutableCollection;
import com.google.common.collect.ImmutableMap;

4
src/java/org/apache/cassandra/fql/FullQueryLogger.java
View File

@ -20,7 +20,6 @@ package org.apache.cassandra.fql;
import java.io.File;
import java.nio.ByteBuffer;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
@ -49,6 +48,7 @@ import org.apache.cassandra.transport.Message;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.binlog.BinLog;
import org.apache.cassandra.utils.binlog.BinLogOptions;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import org.apache.cassandra.utils.concurrent.WeightedQueue;
import org.github.jamm.MemoryLayoutSpecification;
@ -161,7 +161,7 @@ public class FullQueryLogger implements QueryEvents.Listener
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
throw new UncheckedInterruptedException(e);
}
finally
{

1
src/java/org/apache/cassandra/gms/FailureDetector.java
View File

@ -27,7 +27,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.*;
import org.apache.cassandra.locator.Replica;
import org.slf4j.Logger;

11
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -71,6 +71,7 @@ import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.RecomputingSupplier;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static org.apache.cassandra.config.CassandraRelevantProperties.GOSSIPER_QUARANTINE_DELAY;
import static org.apache.cassandra.net.NoPayload.noPayload;
@ -531,7 +532,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
{
task.get();
}
catch (InterruptedException | ExecutionException e)
catch (InterruptedException e)
{
throw new UncheckedInterruptedException(e);
}
catch (ExecutionException e)
{
throw new AssertionError(e);
}
@ -1850,9 +1855,9 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean
}
}
}
catch (InterruptedException wtf)
catch (InterruptedException e)
{
throw new RuntimeException(wtf);
throw new UncheckedInterruptedException(e);
}
return ImmutableMap.copyOf(endpointShadowStateMap);

1
src/java/org/apache/cassandra/hints/ChecksummedDataInput.java
View File

@ -26,7 +26,6 @@ import com.google.common.base.Preconditions;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.util.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.Throwables;
import org.apache.cassandra.utils.NativeLibrary;

8
src/java/org/apache/cassandra/hints/HintsBufferPool.java
View File

@ -20,10 +20,12 @@ package org.apache.cassandra.hints;
import java.io.Closeable;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
/**
* A primitive pool of {@link HintsBuffer} buffers. Under normal conditions should only hold two buffers - the currently
@ -45,7 +47,7 @@ final class HintsBufferPool implements Closeable
HintsBufferPool(int bufferSize, FlushCallback flushCallback)
{
reserveBuffers = new LinkedBlockingQueue<>();
reserveBuffers = newBlockingQueue();
this.bufferSize = bufferSize;
this.flushCallback = flushCallback;
}
@ -117,7 +119,7 @@ final class HintsBufferPool implements Closeable
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
throw new UncheckedInterruptedException(e);
}
}
currentBuffer = buffer == null ? createBuffer() : buffer;

21
src/java/org/apache/cassandra/hints/HintsDispatchExecutor.java
View File

@ -37,6 +37,11 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.FSReadError;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static java.lang.Thread.MIN_PRIORITY;
import static java.util.concurrent.TimeUnit.MINUTES;
import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
/**
* A multi-threaded (by default) executor for dispatching hints.
@ -60,9 +65,9 @@ final class HintsDispatchExecutor
this.isAlive = isAlive;
scheduledDispatches = new ConcurrentHashMap<>();
executor = new JMXEnabledThreadPoolExecutor(maxThreads, 1, TimeUnit.MINUTES,
new LinkedBlockingQueue<>(),
new NamedThreadFactory("HintsDispatcher", Thread.MIN_PRIORITY),
executor = new JMXEnabledThreadPoolExecutor(maxThreads, 1, MINUTES,
newBlockingQueue(),
new NamedThreadFactory("HintsDispatcher", MIN_PRIORITY),
"internal");
}
@ -79,7 +84,7 @@ final class HintsDispatchExecutor
}
catch (InterruptedException e)
{
throw new AssertionError(e);
throw new UncheckedInterruptedException(e);
}
}
@ -120,7 +125,11 @@ final class HintsDispatchExecutor
if (future != null)
future.get();
}
catch (ExecutionException | InterruptedException e)
catch (InterruptedException e)
{
throw new UncheckedInterruptedException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
@ -167,7 +176,7 @@ final class HintsDispatchExecutor
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
throw new UncheckedInterruptedException(e);
}
hostId = hostIdSupplier.get();

3
src/java/org/apache/cassandra/hints/HintsDispatchTrigger.java
View File

@ -19,12 +19,9 @@ package org.apache.cassandra.hints;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.cassandra.gms.ApplicationState;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.schema.Schema;
import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
/**
* A simple dispatch trigger that's being run every 10 seconds.
*

20
src/java/org/apache/cassandra/hints/HintsDispatcher.java
View File

@ -24,6 +24,7 @@ import java.util.function.BooleanSupplier;
import java.util.function.Function;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.cassandra.utils.concurrent.Condition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -33,10 +34,13 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.metrics.HintsServiceMetrics;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.concurrent.SimpleCondition;
import static org.apache.cassandra.hints.HintsDispatcher.Callback.Outcome.*;
import static org.apache.cassandra.metrics.HintsServiceMetrics.updateDelayMetrics;
import static org.apache.cassandra.net.Verb.HINT_REQ;
import static org.apache.cassandra.utils.MonotonicClock.approxTime;
import static org.apache.cassandra.utils.concurrent.Condition.newOneTimeCondition;
/**
* Dispatches a single hints file to a specified node in a batched manner.
@ -205,12 +209,12 @@ final class HintsDispatcher implements AutoCloseable
return callback;
}
private static final class Callback implements RequestCallback
static final class Callback implements RequestCallback
{
enum Outcome { SUCCESS, TIMEOUT, FAILURE, INTERRUPTED }
private final long start = approxTime.now();
private final SimpleCondition condition = new SimpleCondition();
private final Condition condition = newOneTimeCondition();
private volatile Outcome outcome;
private final long hintCreationNanoTime;
@ -229,10 +233,10 @@ final class HintsDispatcher implements AutoCloseable
catch (InterruptedException e)
{
logger.warn("Hint dispatch was interrupted", e);
return Outcome.INTERRUPTED;
return INTERRUPTED;
}
return timedOut ? Outcome.TIMEOUT : outcome;
return timedOut ? TIMEOUT : outcome;
}
@Override
@ -244,15 +248,15 @@ final class HintsDispatcher implements AutoCloseable
@Override
public void onFailure(InetAddressAndPort from, RequestFailureReason failureReason)
{
outcome = Outcome.FAILURE;
outcome = FAILURE;
condition.signalAll();
}
@Override
public void onResponse(Message msg)
{
HintsServiceMetrics.updateDelayMetrics(msg.from(), approxTime.now() - this.hintCreationNanoTime);
outcome = Outcome.SUCCESS;
updateDelayMetrics(msg.from(), approxTime.now() - this.hintCreationNanoTime);
outcome = SUCCESS;
condition.signalAll();
}
}

13
src/java/org/apache/cassandra/hints/HintsService.java
View File

@ -48,6 +48,7 @@ import org.apache.cassandra.dht.Token;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.MBeanWrapper;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static com.google.common.collect.Iterables.transform;
@ -339,7 +340,11 @@ public final class HintsService implements HintsServiceMBean
flushFuture.get();
closeFuture.get();
}
catch (InterruptedException | ExecutionException e)
catch (InterruptedException e)
{
throw new UncheckedInterruptedException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
@ -376,7 +381,11 @@ public final class HintsService implements HintsServiceMBean
flushFuture.get();
closeFuture.get();
}
catch (InterruptedException | ExecutionException e)
catch (InterruptedException e)
{
throw new UncheckedInterruptedException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}

7
src/java/org/apache/cassandra/hints/HintsWriteExecutor.java
View File

@ -30,6 +30,7 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.FSError;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
/**
* A single threaded executor that exclusively writes all the hints and otherwise manipulate the writers.
@ -102,7 +103,11 @@ final class HintsWriteExecutor
{
executor.submit(new FsyncWritersTask(stores)).get();
}
catch (InterruptedException | ExecutionException e)
catch (InterruptedException e)
{
throw new UncheckedInterruptedException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}

10
src/java/org/apache/cassandra/index/SecondaryIndexManager.java
View File

@ -48,7 +48,6 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.statements.schema.IndexTarget;
import org.apache.cassandra.db.*;
@ -79,8 +78,11 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.concurrent.Refs;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.cassandra.concurrent.Stage.KEEP_ALIVE_SECONDS;
import static org.apache.cassandra.utils.ExecutorUtils.awaitTermination;
import static org.apache.cassandra.utils.ExecutorUtils.shutdown;
import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
/**
* Handles the core maintenance functionality associated with indexes: adding/removing them to or from
@ -162,9 +164,9 @@ public class SecondaryIndexManager implements IndexRegistry, INotificationConsum
// executes tasks returned by Indexer#addIndexColumn which may require index(es) to be (re)built
private static final ListeningExecutorService asyncExecutor = MoreExecutors.listeningDecorator(
new JMXEnabledThreadPoolExecutor(1,
Stage.KEEP_ALIVE_SECONDS,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(),
KEEP_ALIVE_SECONDS,
SECONDS,
newBlockingQueue(),
new NamedThreadFactory("SecondaryIndexManagement"),
"internal"));

2
src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java
View File

@ -24,7 +24,6 @@ import java.util.List;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.ClusteringIndexNamesFilter;
import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
@ -34,7 +33,6 @@ import org.apache.cassandra.index.internal.CassandraIndex;
import org.apache.cassandra.index.internal.CassandraIndexSearcher;
import org.apache.cassandra.index.internal.IndexEntry;
import org.apache.cassandra.utils.btree.BTreeSet;
import org.apache.cassandra.utils.concurrent.OpOrder;
public class CompositesSearcher extends CassandraIndexSearcher

24
src/java/org/apache/cassandra/index/sasi/TermIterator.java
View File

@ -26,7 +26,6 @@ import java.util.concurrent.atomic.AtomicLong;
import io.netty.util.concurrent.FastThreadLocal;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.index.sasi.disk.OnDiskIndexBuilder;
import org.apache.cassandra.index.sasi.disk.Token;
import org.apache.cassandra.index.sasi.plan.Expression;
import org.apache.cassandra.index.sasi.utils.RangeUnionIterator;
@ -34,11 +33,16 @@ import org.apache.cassandra.index.sasi.utils.RangeIterator;
import org.apache.cassandra.io.util.FileUtils;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.utils.concurrent.CountDownLatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static java.lang.String.format;
import static org.apache.cassandra.index.sasi.disk.OnDiskIndexBuilder.Mode.CONTAINS;
import static org.apache.cassandra.index.sasi.plan.Expression.Op.PREFIX;
import static org.apache.cassandra.utils.concurrent.CountDownLatch.newCountDownLatch;
public class TermIterator extends RangeIterator<Long, Token>
{
private static final Logger logger = LoggerFactory.getLogger(TermIterator.class);
@ -99,14 +103,14 @@ public class TermIterator extends RangeIterator<Long, Token>
try
{
final CountDownLatch latch = new CountDownLatch(perSSTableIndexes.size());
final CountDownLatch latch = newCountDownLatch(perSSTableIndexes.size());
final ExecutorService searchExecutor = SEARCH_EXECUTOR.get();
for (final SSTableIndex index : perSSTableIndexes)
{
if (e.getOp() == Expression.Op.PREFIX &&
index.mode() == OnDiskIndexBuilder.Mode.CONTAINS && !index.hasMarkedPartials())
throw new UnsupportedOperationException(String.format("The index %s has not yet been upgraded " +
if (e.getOp() == PREFIX &&
index.mode() == CONTAINS && !index.hasMarkedPartials())
throw new UnsupportedOperationException(format("The index %s has not yet been upgraded " +
"to support prefix queries in CONTAINS mode. " +
"Wait for compaction or rebuild the index.",
index.getPath()));
@ -114,7 +118,7 @@ public class TermIterator extends RangeIterator<Long, Token>
if (!index.reference())
{
latch.countDown();
latch.decrement();
continue;
}
@ -142,16 +146,16 @@ public class TermIterator extends RangeIterator<Long, Token>
releaseIndex(referencedIndexes, index);
if (logger.isDebugEnabled())
logger.debug(String.format("Failed search an index %s, skipping.", index.getPath()), e1);
logger.debug(format("Failed search an index %s, skipping.", index.getPath()), e1);
}
finally
{
latch.countDown();
latch.decrement();
}
});
}
Uninterruptibles.awaitUninterruptibly(latch);
latch.awaitUninterruptibly();
// checkpoint right away after all indexes complete search because we might have crossed the quota
e.checkpoint();

20
src/java/org/apache/cassandra/index/sasi/disk/PerSSTableIndexWriter.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.index.sasi.disk;
import java.io.File;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
@ -43,16 +42,19 @@ import org.apache.cassandra.io.sstable.format.SSTableFlushObserver;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.CountDownLatch;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.Uninterruptibles;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static java.util.concurrent.TimeUnit.MINUTES;
import static org.apache.cassandra.utils.concurrent.CountDownLatch.newCountDownLatch;
import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
public class PerSSTableIndexWriter implements SSTableFlushObserver
{
@ -64,14 +66,14 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
static
{
INDEX_FLUSHER_GENERAL = new JMXEnabledThreadPoolExecutor(POOL_SIZE, POOL_SIZE, 1, TimeUnit.MINUTES,
new LinkedBlockingQueue<>(),
INDEX_FLUSHER_GENERAL = new JMXEnabledThreadPoolExecutor(POOL_SIZE, POOL_SIZE, 1, MINUTES,
newBlockingQueue(),
new NamedThreadFactory("SASI-General"),
"internal");
INDEX_FLUSHER_GENERAL.allowCoreThreadTimeOut(true);
INDEX_FLUSHER_MEMTABLE = new JMXEnabledThreadPoolExecutor(POOL_SIZE, POOL_SIZE, 1, TimeUnit.MINUTES,
new LinkedBlockingQueue<>(),
INDEX_FLUSHER_MEMTABLE = new JMXEnabledThreadPoolExecutor(POOL_SIZE, POOL_SIZE, 1, MINUTES,
newBlockingQueue(),
new NamedThreadFactory("SASI-Memtable"),
"internal");
INDEX_FLUSHER_MEMTABLE.allowCoreThreadTimeOut(true);
@ -141,11 +143,11 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
try
{
CountDownLatch latch = new CountDownLatch(indexes.size());
CountDownLatch latch = newCountDownLatch(indexes.size());
for (Index index : indexes.values())
index.complete(latch);
Uninterruptibles.awaitUninterruptibly(latch);
latch.awaitUninterruptibly();
}
finally
{
@ -339,7 +341,7 @@ public class PerSSTableIndexWriter implements SSTableFlushObserver
FileUtils.delete(outputFile + "_" + segment);
}
latch.countDown();
latch.decrement();
}
});
}

2
src/java/org/apache/cassandra/io/compress/CompressionMetadata.java
View File

@ -26,12 +26,10 @@ import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Map;
import java.util.SortedSet;

4
src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
View File

@ -36,6 +36,8 @@ import org.apache.cassandra.db.rows.UnfilteredSerializer;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
/**
* A SSTable writer that doesn't assume rows are in sorted order.
@ -165,7 +167,7 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
throw new UncheckedInterruptedException(e);
}
}
}

10
src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
View File

@ -69,12 +69,10 @@ import org.apache.cassandra.schema.TableMetadataRef;
import org.apache.cassandra.service.ActiveRepairService;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.concurrent.SelfRefCounted;
import org.apache.cassandra.utils.BloomFilterSerializer;
import org.apache.cassandra.utils.concurrent.*;
import static org.apache.cassandra.db.Directories.SECONDARY_INDEX_NAME_SEPARATOR;
import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
/**
* An SSTableReader can be constructed in a number of places, but typically is either
@ -525,7 +523,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
public static Collection<SSTableReader> openAll(Set<Map.Entry<Descriptor, Set<Component>>> entries,
final TableMetadataRef metadata)
{
final Collection<SSTableReader> sstables = new LinkedBlockingQueue<>();
final Collection<SSTableReader> sstables = newBlockingQueue();
ExecutorService executor = DebuggableThreadPoolExecutor.createWithFixedPoolSize("SSTableBatchOpen", FBUtilities.getAvailableProcessors());
for (final Map.Entry<Descriptor, Set<Component>> entry : entries)
@ -564,7 +562,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
}
catch (InterruptedException e)
{
throw new AssertionError(e);
throw new UncheckedInterruptedException(e);
}
return sstables;

1
src/java/org/apache/cassandra/io/util/BufferedDataOutputStreamPlus.java
View File

@ -29,7 +29,6 @@ import com.google.common.base.Preconditions;
import net.nicoulaj.compilecommand.annotations.DontInline;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.utils.FastByteOperations;
import org.apache.cassandra.utils.memory.MemoryUtil;
/**
* An implementation of the DataOutputStreamPlus interface using a ByteBuffer to stage writes

2
src/java/org/apache/cassandra/io/util/UnbufferedDataOutputStreamPlus.java
View File

@ -25,8 +25,6 @@ import java.nio.channels.WritableByteChannel;
import org.apache.cassandra.utils.memory.MemoryUtil;
import com.google.common.base.Function;
/**
* Base class for DataOutput implementations that does not have an optimized implementations of Plus methods
* and does no buffering.

2
src/java/org/apache/cassandra/locator/Endpoints.java
View File

@ -28,8 +28,6 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
import com.google.common.collect.Lists;
/**
* A collection of Endpoints for a given ring position. This will typically reside in a ReplicaLayout,
* representing some subset of the endpoints for the Token or Range

2
src/java/org/apache/cassandra/locator/EndpointsByReplica.java
View File

@ -23,8 +23,6 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import org.apache.cassandra.locator.ReplicaCollection.Builder.Conflict;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
public class EndpointsByReplica extends ReplicaMultimap<Replica, EndpointsForRange>

2
src/java/org/apache/cassandra/locator/InOurDcTester.java
View File

@ -19,7 +19,7 @@
package org.apache.cassandra.locator;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.utils.FBUtilities;
import java.util.function.Predicate;
public class InOurDcTester

2
src/java/org/apache/cassandra/locator/RangesByEndpoint.java
View File

@ -22,8 +22,6 @@ import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
public class RangesByEndpoint extends ReplicaMultimap<InetAddressAndPort, RangesAtEndpoint>

2
src/java/org/apache/cassandra/locator/ReconnectableSnitchHelper.java
View File

@ -30,8 +30,6 @@ import org.apache.cassandra.net.OutboundConnectionSettings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.apache.cassandra.net.ConnectionType.SMALL_MESSAGES;
/**
* Sidekick helper for snitches that want to reconnect from one IP addr for a node to another.
* Typically, this is for situations like EC2 where a node will have a public address and a private address,

1
src/java/org/apache/cassandra/locator/ReplicaPlans.java
View File

@ -19,7 +19,6 @@
package org.apache.cassandra.locator;
import com.carrotsearch.hppc.ObjectIntHashMap;
import com.carrotsearch.hppc.cursors.ObjectIntCursor;
import com.carrotsearch.hppc.cursors.ObjectObjectCursor;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ArrayListMultimap;

1
src/java/org/apache/cassandra/locator/TokenMetadata.java
View File

@ -30,6 +30,7 @@ import javax.annotation.concurrent.GuardedBy;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.*;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

2
src/java/org/apache/cassandra/metrics/CompactionMetrics.java
View File

@ -26,10 +26,8 @@ import com.codahale.metrics.Meter;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.compaction.ActiveCompactions;
import org.apache.cassandra.db.compaction.CompactionInfo;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;

27
src/java/org/apache/cassandra/net/AsyncChannelPromise.java
View File

@ -24,6 +24,7 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
/**
* See {@link AsyncPromise} and {@link io.netty.channel.ChannelPromise}
@ -92,7 +93,7 @@ public class AsyncChannelPromise extends AsyncPromise<Void> implements ChannelPr
return setSuccess(null);
}
public ChannelPromise setSuccess(Void v)
public AsyncChannelPromise setSuccess(Void v)
{
super.setSuccess(v);
return this;
@ -103,58 +104,56 @@ public class AsyncChannelPromise extends AsyncPromise<Void> implements ChannelPr
return trySuccess(null);
}
public ChannelPromise setFailure(Throwable throwable)
public AsyncChannelPromise setFailure(Throwable throwable)
{
super.setFailure(throwable);
return this;
}
public ChannelPromise sync() throws InterruptedException
public AsyncChannelPromise sync() throws InterruptedException
{
super.sync();
return this;
}
public ChannelPromise syncUninterruptibly()
public AsyncChannelPromise syncUninterruptibly()
{
super.syncUninterruptibly();
return this;
}
public ChannelPromise await() throws InterruptedException
public AsyncChannelPromise await() throws InterruptedException
{
super.await();
return this;
}
public ChannelPromise awaitUninterruptibly()
public AsyncChannelPromise awaitUninterruptibly()
{
super.awaitUninterruptibly();
return this;
}
public ChannelPromise addListener(GenericFutureListener<? extends Future<? super Void>> listener)
public AsyncChannelPromise addListener(GenericFutureListener<? extends Future<? super Void>> listener)
{
super.addListener(listener);
return this;
}
public ChannelPromise addListeners(GenericFutureListener<? extends Future<? super Void>>... listeners)
public AsyncChannelPromise addListeners(GenericFutureListener<? extends Future<? super Void>>... listeners)
{
super.addListeners(listeners);
return this;
}
public ChannelPromise removeListener(GenericFutureListener<? extends Future<? super Void>> listener)
public AsyncChannelPromise removeListener(GenericFutureListener<? extends Future<? super Void>> listener)
{
super.removeListener(listener);
return this;
throw new UnsupportedOperationException();
}
public ChannelPromise removeListeners(GenericFutureListener<? extends Future<? super Void>>... listeners)
public AsyncChannelPromise removeListeners(GenericFutureListener<? extends Future<? super Void>>... listeners)
{
super.removeListeners(listeners);
return this;
throw new UnsupportedOperationException();
}
public ChannelPromise unvoid()

7
src/java/org/apache/cassandra/net/AsyncOneResponse.java
View File

@ -19,7 +19,7 @@ package org.apache.cassandra.net;
import com.google.common.annotations.VisibleForTesting;
import io.netty.util.concurrent.ImmediateEventExecutor;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
/**
* A callback specialized for returning a value from a single target; that is, this is for messages
@ -27,11 +27,6 @@ import io.netty.util.concurrent.ImmediateEventExecutor;
*/
public class AsyncOneResponse<T> extends AsyncPromise<T> implements RequestCallback<T>
{
public AsyncOneResponse()
{
super(ImmediateEventExecutor.INSTANCE);
}
public void onResponse(Message<T> response)
{
setSuccess(response.payload);

489
src/java/org/apache/cassandra/net/AsyncPromise.java
View File

@ -1,489 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.net;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.Promise;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.ThrowableUtil;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static java.util.concurrent.atomic.AtomicReferenceFieldUpdater.*;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* Netty's DefaultPromise uses a mutex to coordinate notifiers AND waiters between the eventLoop and the other threads.
* Since we register cross-thread listeners, this has the potential to block internode messaging for an unknown
* number of threads for an unknown period of time, if we are unlucky with the scheduler (which will certainly
* happen, just with some unknown but low periodicity)
*
* At the same time, we manage some other efficiencies:
* - We save some space when registering listeners, especially if there is only one listener, as we perform no
* extra allocations in this case.
* - We permit efficient initial state declaration, avoiding unnecessary CAS or lock acquisitions when mutating
* a Promise we are ourselves constructing (and can easily add more; only those we use have been added)
*
* We can also make some guarantees about our behaviour here, although we primarily mirror Netty.
* Specifically, we can guarantee that notifiers are always invoked in the order they are added (which may be true
* for netty, but was unclear and is not declared). This is useful for ensuring the correctness of some of our
* behaviours in OutboundConnection without having to jump through extra hoops.
*
* The implementation loosely follows that of Netty's DefaultPromise, with some slight changes; notably that we have
* no synchronisation on our listeners, instead using a CoW list that is cleared each time we notify listeners.
*
* We handle special values slightly differently. We do not use a special value for null, instead using
* a special value to indicate the result has not been set yet. This means that once isSuccess() holds,
* the result must be a correctly typed object (modulo generics pitfalls).
* All special values are also instances of FailureHolder, which simplifies a number of the logical conditions.
*
* @param <V>
*/
public class AsyncPromise<V> implements Promise<V>
{
private static final Logger logger = LoggerFactory.getLogger(AsyncPromise.class);
private final EventExecutor executor;
private volatile Object result;
private volatile GenericFutureListener<? extends Future<? super V>> listeners;
private volatile WaitQueue waiting;
private static final AtomicReferenceFieldUpdater<AsyncPromise, Object> resultUpdater = newUpdater(AsyncPromise.class, Object.class, "result");
private static final AtomicReferenceFieldUpdater<AsyncPromise, GenericFutureListener> listenersUpdater = newUpdater(AsyncPromise.class, GenericFutureListener.class, "listeners");
private static final AtomicReferenceFieldUpdater<AsyncPromise, WaitQueue> waitingUpdater = newUpdater(AsyncPromise.class, WaitQueue.class, "waiting");
private static final FailureHolder UNSET = new FailureHolder(null);
private static final FailureHolder UNCANCELLABLE = new FailureHolder(null);
private static final FailureHolder CANCELLED = new FailureHolder(ThrowableUtil.unknownStackTrace(new CancellationException(), AsyncPromise.class, "cancel(...)"));
private static final DeferredGenericFutureListener NOTIFYING = future -> {};
private static interface DeferredGenericFutureListener<F extends Future<?>> extends GenericFutureListener<F> {}
private static final class FailureHolder
{
final Throwable cause;
private FailureHolder(Throwable cause)
{
this.cause = cause;
}
}
public AsyncPromise(EventExecutor executor)
{
this(executor, UNSET);
}
private AsyncPromise(EventExecutor executor, FailureHolder initialState)
{
this.executor = executor;
this.result = initialState;
}
public AsyncPromise(EventExecutor executor, GenericFutureListener<? extends Future<? super V>> listener)
{
this(executor);
this.listeners = listener;
}
AsyncPromise(EventExecutor executor, FailureHolder initialState, GenericFutureListener<? extends Future<? super V>> listener)
{
this(executor, initialState);
this.listeners = listener;
}
public static <V> AsyncPromise<V> uncancellable(EventExecutor executor)
{
return new AsyncPromise<>(executor, UNCANCELLABLE);
}
public static <V> AsyncPromise<V> uncancellable(EventExecutor executor, GenericFutureListener<? extends Future<? super V>> listener)
{
return new AsyncPromise<>(executor, UNCANCELLABLE);
}
public Promise<V> setSuccess(V v)
{
if (!trySuccess(v))
throw new IllegalStateException("complete already: " + this);
return this;
}
public Promise<V> setFailure(Throwable throwable)
{
if (!tryFailure(throwable))
throw new IllegalStateException("complete already: " + this);
return this;
}
public boolean trySuccess(V v)
{
return trySet(v);
}
public boolean tryFailure(Throwable throwable)
{
return trySet(new FailureHolder(throwable));
}
public boolean setUncancellable()
{
if (trySet(UNCANCELLABLE))
return true;
return result == UNCANCELLABLE;
}
public boolean cancel(boolean b)
{
return trySet(CANCELLED);
}
/**
* Shared implementation of various promise completion methods.
* Updates the result if it is possible to do so, returning success/failure.
*
* If the promise is UNSET the new value will succeed;
* if it is UNCANCELLABLE it will succeed only if the new value is not CANCELLED
* otherwise it will fail, as isDone() is implied
*
* If the update succeeds, and the new state implies isDone(), any listeners and waiters will be notified
*/
private boolean trySet(Object v)
{
while (true)
{
Object current = result;
if (isDone(current) || (current == UNCANCELLABLE && v == CANCELLED))
return false;
if (resultUpdater.compareAndSet(this, current, v))
{
if (v != UNCANCELLABLE)
{
notifyListeners();
notifyWaiters();
}
return true;
}
}
}
public boolean isSuccess()
{
return isSuccess(result);
}
private static boolean isSuccess(Object result)
{
return !(result instanceof FailureHolder);
}
public boolean isCancelled()
{
return isCancelled(result);
}
private static boolean isCancelled(Object result)
{
return result == CANCELLED;
}
public boolean isDone()
{
return isDone(result);
}
private static boolean isDone(Object result)
{
return result != UNSET && result != UNCANCELLABLE;
}
public boolean isCancellable()
{
Object result = this.result;
return result == UNSET;
}
public Throwable cause()
{
Object result = this.result;
if (result instanceof FailureHolder)
return ((FailureHolder) result).cause;
return null;
}
/**
* if isSuccess(), returns the value, otherwise returns null
*/
@SuppressWarnings("unchecked")
public V getNow()
{
Object result = this.result;
if (isSuccess(result))
return (V) result;
return null;
}
public V get() throws InterruptedException, ExecutionException
{
await();
return getWhenDone();
}
public V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException
{
if (!await(timeout, unit))
throw new TimeoutException();
return getWhenDone();
}
/**
* Shared implementation of get() after suitable await(); assumes isDone(), and returns
* either the success result or throws the suitable exception under failure
*/
@SuppressWarnings("unchecked")
private V getWhenDone() throws ExecutionException
{
Object result = this.result;
if (isSuccess(result))
return (V) result;
if (result == CANCELLED)
throw new CancellationException();
throw new ExecutionException(((FailureHolder) result).cause);
}
/**
* waits for completion; in case of failure rethrows the original exception without a new wrapping exception
* so may cause problems for reporting stack traces
*/
public Promise<V> sync() throws InterruptedException
{
await();
rethrowIfFailed();
return this;
}
/**
* waits for completion; in case of failure rethrows the original exception without a new wrapping exception
* so may cause problems for reporting stack traces
*/
public Promise<V> syncUninterruptibly()
{
awaitUninterruptibly();
rethrowIfFailed();
return this;
}
private void rethrowIfFailed()
{
Throwable cause = this.cause();
if (cause != null)
{
PlatformDependent.throwException(cause);
}
}
public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener)
{
listenersUpdater.accumulateAndGet(this, listener, AsyncPromise::appendListener);
if (isDone())
notifyListeners();
return this;
}
public Promise<V> addListeners(GenericFutureListener<? extends Future<? super V>> ... listeners)
{
// this could be more efficient if we cared, but we do not
return addListener(future -> {
for (GenericFutureListener<? extends Future<? super V>> listener : listeners)
AsyncPromise.invokeListener((GenericFutureListener<Future<? super V>>)listener, future);
});
}
public Promise<V> removeListener(GenericFutureListener<? extends Future<? super V>> listener)
{
throw new UnsupportedOperationException();
}
public Promise<V> removeListeners(GenericFutureListener<? extends Future<? super V>> ... listeners)
{
throw new UnsupportedOperationException();
}
@SuppressWarnings("unchecked")
private void notifyListeners()
{
if (!executor.inEventLoop())
{
// submit this method, to guarantee we invoke in the submitted order
executor.execute(this::notifyListeners);
return;
}
if (listeners == null || listeners instanceof DeferredGenericFutureListener<?>)
return; // either no listeners, or we are already notifying listeners, so we'll get to the new one when ready
// first run our notifiers
while (true)
{
GenericFutureListener listeners = listenersUpdater.getAndSet(this, NOTIFYING);
if (listeners != null)
invokeListener(listeners, this);
if (listenersUpdater.compareAndSet(this, NOTIFYING, null))
return;
}
}
private static <F extends Future<?>> void invokeListener(GenericFutureListener<F> listener, F future)
{
try
{
listener.operationComplete(future);
}
catch (Throwable t)
{
logger.error("Failed to invoke listener {} to {}", listener, future, t);
}
}
private static <F extends Future<?>> GenericFutureListener<F> appendListener(GenericFutureListener<F> prevListener, GenericFutureListener<F> newListener)
{
GenericFutureListener<F> result = newListener;
if (prevListener != null && prevListener != NOTIFYING)
{
result = future -> {
invokeListener(prevListener, future);
// we will wrap the outer invocation with invokeListener, so no need to do it here too
newListener.operationComplete(future);
};
}
if (prevListener instanceof DeferredGenericFutureListener<?>)
{
GenericFutureListener<F> wrap = result;
result = (DeferredGenericFutureListener<F>) wrap::operationComplete;
}
return result;
}
public Promise<V> await() throws InterruptedException
{
await(0L, (signal, nanos) -> { signal.await(); return true; } );
return this;
}
public Promise<V> awaitUninterruptibly()
{
await(0L, (signal, nanos) -> { signal.awaitUninterruptibly(); return true; } );
return this;
}
public boolean await(long timeout, TimeUnit unit) throws InterruptedException
{
return await(unit.toNanos(timeout),
(signal, nanos) -> signal.awaitUntil(nanos + nanoTime()));
}
public boolean await(long timeoutMillis) throws InterruptedException
{
return await(timeoutMillis, TimeUnit.MILLISECONDS);
}
public boolean awaitUninterruptibly(long timeout, TimeUnit unit)
{
return await(unit.toNanos(timeout),
(signal, nanos) -> signal.awaitUntilUninterruptibly(nanos + nanoTime()));
}
public boolean awaitUninterruptibly(long timeoutMillis)
{
return awaitUninterruptibly(timeoutMillis, TimeUnit.MILLISECONDS);
}
interface Awaiter<T extends Throwable>
{
boolean await(WaitQueue.Signal value, long nanos) throws T;
}
/**
* A clean way to implement each variant of await using lambdas; we permit a nanos parameter
* so that we can implement this without any unnecessary lambda allocations, although not
* all implementations need the nanos parameter (i.e. those that wait indefinitely)
*/
private <T extends Throwable> boolean await(long nanos, Awaiter<T> awaiter) throws T
{
if (isDone())
return true;
WaitQueue.Signal await = registerToWait();
if (null != await)
return awaiter.await(await, nanos);
return true;
}
/**
* Register a signal that will be notified when the promise is completed;
* if the promise becomes completed before this signal is registered, null is returned
*/
private WaitQueue.Signal registerToWait()
{
WaitQueue waiting = this.waiting;
if (waiting == null && !waitingUpdater.compareAndSet(this, null, waiting = new WaitQueue()))
waiting = this.waiting;
assert waiting != null;
WaitQueue.Signal signal = waiting.register();
if (!isDone())
return signal;
signal.cancel();
return null;
}
private void notifyWaiters()
{
WaitQueue waiting = this.waiting;
if (waiting != null)
waiting.signalAll();
}
public String toString()
{
Object result = this.result;
if (isSuccess(result))
return "(success: " + result + ')';
if (result == UNCANCELLABLE)
return "(uncancellable)";
if (result == CANCELLED)
return "(cancelled)";
if (isDone(result))
return "(failure: " + ((FailureHolder) result).cause + ')';
return "(incomplete)";
}
}

5
src/java/org/apache/cassandra/net/AsyncStreamingInputPlus.java
View File

@ -21,7 +21,6 @@ import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import com.google.common.annotations.VisibleForTesting;
@ -33,6 +32,8 @@ import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import org.apache.cassandra.io.util.RebufferingInputStream;
import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
// TODO: rewrite
public class AsyncStreamingInputPlus extends RebufferingInputStream
{
@ -65,7 +66,7 @@ public class AsyncStreamingInputPlus extends RebufferingInputStream
super(Unpooled.EMPTY_BUFFER.nioBuffer());
currentBuf = Unpooled.EMPTY_BUFFER;
queue = new LinkedBlockingQueue<>();
queue = newBlockingQueue();
rebufferTimeoutNanos = rebufferTimeoutUnit.toNanos(rebufferTimeout);
this.channel = channel;

1
src/java/org/apache/cassandra/net/FrameDecoderCrc.java
View File

@ -25,7 +25,6 @@ import java.util.zip.CRC32;
import io.netty.channel.ChannelPipeline;
import static org.apache.cassandra.net.Crc.*;
import static org.apache.cassandra.net.Crc.updateCrc32;
/**
* Framing format that protects integrity of data in movement with CRCs (of both header and payload).

1
src/java/org/apache/cassandra/net/FrameEncoderCrc.java
View File

@ -23,7 +23,6 @@ import java.util.zip.CRC32;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import org.apache.cassandra.utils.memory.BufferPool;
import static org.apache.cassandra.net.Crc.*;

1
src/java/org/apache/cassandra/net/FrameEncoderLegacyLZ4.java
View File

@ -27,7 +27,6 @@ import net.jpountz.xxhash.XXHash32;
import net.jpountz.xxhash.XXHashFactory;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.memory.BufferPool;
import static java.lang.Integer.reverseBytes;
import static java.lang.Math.min;

1
src/java/org/apache/cassandra/net/FrameEncoderUnprotected.java
View File

@ -21,7 +21,6 @@ import java.nio.ByteBuffer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import org.apache.cassandra.utils.memory.BufferPool;
import static org.apache.cassandra.net.FrameEncoderCrc.HEADER_LENGTH;
import static org.apache.cassandra.net.FrameEncoderCrc.writeHeader;

1
src/java/org/apache/cassandra/net/FutureCombiner.java
View File

@ -25,6 +25,7 @@ import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.GlobalEventExecutor;
import io.netty.util.concurrent.Promise;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
/**
* Netty's PromiseCombiner is not threadsafe, and we combine futures from multiple event executors.

4
src/java/org/apache/cassandra/net/HandshakeProtocol.java
View File

@ -40,7 +40,6 @@ import static org.apache.cassandra.net.MessagingService.VERSION_30;
import static org.apache.cassandra.net.MessagingService.VERSION_40;
import static org.apache.cassandra.net.Message.validateLegacyProtocolMagic;
import static org.apache.cassandra.net.Crc.*;
import static org.apache.cassandra.net.Crc.computeCrc32;
import static org.apache.cassandra.net.OutboundConnectionSettings.*;
/**
@ -53,7 +52,8 @@ import static org.apache.cassandra.net.OutboundConnectionSettings.*;
* it will simply disconnect and reconnect with a more appropriate version. But if the version is acceptable, the connection
* initiator sends the third message of the protocol, after which it considers the connection ready.
*/
class HandshakeProtocol
@VisibleForTesting
public class HandshakeProtocol
{
static final long TIMEOUT_MILLIS = 3 * DatabaseDescriptor.getRpcTimeout(MILLISECONDS);

3
src/java/org/apache/cassandra/net/InboundConnectionInitiator.java
View File

@ -58,9 +58,6 @@ import org.apache.cassandra.utils.memory.BufferPools;
import static java.lang.Math.*;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.cassandra.net.MessagingService.*;
import static org.apache.cassandra.net.MessagingService.VERSION_40;
import static org.apache.cassandra.net.MessagingService.current_version;
import static org.apache.cassandra.net.MessagingService.minimum_version;
import static org.apache.cassandra.net.SocketFactory.WIRETRACE;
import static org.apache.cassandra.net.SocketFactory.newSslHandler;

1
src/java/org/apache/cassandra/net/InboundSockets.java
View File

@ -38,6 +38,7 @@ import io.netty.util.concurrent.SucceededFuture;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
class InboundSockets
{

1
src/java/org/apache/cassandra/net/MessagingServiceMBeanImpl.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.net;
import java.io.IOException;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

Some files were not shown because too many files have changed in this diff Show More