diff --git a/CHANGES.txt b/CHANGES.txt index 380181c4cc..1f97f2f475 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -15,6 +15,7 @@ * Add optional keyspace to DROP INDEX statement (CASSANDRA-7314) * Reduce run time for CQL tests (CASSANDRA-7327) * Fix heap size calculation on Windows (CASSANDRA-7352) + * RefCount native frames from netty (CASSANDRA-7245) Merged from 2.0: * Add per-CF range read request latency metrics (CASSANDRA-7338) * Fix NPE in StreamTransferTask.createMessageForRetry() (CASSANDRA-7323) diff --git a/build.xml b/build.xml index df5514a4be..2e214e3f8c 100644 --- a/build.xml +++ b/build.xml @@ -378,7 +378,7 @@ - + diff --git a/lib/netty-all-4.0.17.Final.jar b/lib/netty-all-4.0.19.Final.jar similarity index 65% rename from lib/netty-all-4.0.17.Final.jar rename to lib/netty-all-4.0.19.Final.jar index baaa5b83c1..66d58f3ffe 100644 Binary files a/lib/netty-all-4.0.17.Final.jar and b/lib/netty-all-4.0.19.Final.jar differ diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index a302af0ebf..546cf3ba53 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -22,6 +22,7 @@ import java.util.*; import com.google.common.base.Function; import com.google.common.collect.Iterables; +import org.apache.cassandra.transport.Frame; import org.github.jamm.MemoryMeter; import org.apache.cassandra.auth.Permission; @@ -495,7 +496,7 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF else cl.validateForWrite(cfm.ksName); - Collection mutations = getMutations(options, false, options.getTimestamp(queryState)); + Collection mutations = getMutations(options, false, options.getTimestamp(queryState), queryState.getSourceFrame()); if (!mutations.isEmpty()) StorageProxy.mutateWithTriggers(mutations, cl, false); @@ -633,10 +634,11 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF if (hasConditions()) throw new UnsupportedOperationException(); - for (IMutation mutation : getMutations(options, true, queryState.getTimestamp())) + for (IMutation mutation : getMutations(options, true, queryState.getTimestamp(), queryState.getSourceFrame())) { // We don't use counters internally. assert mutation instanceof Mutation; + ((Mutation) mutation).apply(); } return null; @@ -652,7 +654,7 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF * @return list of the mutations * @throws InvalidRequestException on invalid requests */ - private Collection getMutations(QueryOptions options, boolean local, long now) + private Collection getMutations(QueryOptions options, boolean local, long now, Frame sourceFrame) throws RequestExecutionException, RequestValidationException { List keys = buildPartitionKeyNames(options); @@ -660,13 +662,15 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF UpdateParameters params = makeUpdateParameters(keys, clusteringPrefix, options, local, now); - Collection mutations = new ArrayList(); + Collection mutations = new ArrayList(keys.size()); for (ByteBuffer key: keys) { ThriftValidation.validateKey(cfm, key); ColumnFamily cf = ArrayBackedSortedColumns.factory.create(cfm); addUpdateForKey(cf, key, clusteringPrefix, params); Mutation mut = new Mutation(cfm.ksName, key, cf); + mut.setSourceFrame(sourceFrame); + mutations.add(isCounter() ? new CounterMutation(mut, options.getConsistency()) : mut); } return mutations; diff --git a/src/java/org/apache/cassandra/db/CounterMutation.java b/src/java/org/apache/cassandra/db/CounterMutation.java index 58889c17d8..95f4ce3ef2 100644 --- a/src/java/org/apache/cassandra/db/CounterMutation.java +++ b/src/java/org/apache/cassandra/db/CounterMutation.java @@ -69,6 +69,18 @@ public class CounterMutation implements IMutation return mutation.getColumnFamilies(); } + @Override + public void retain() + { + mutation.retain(); + } + + @Override + public void release() + { + mutation.release(); + } + public Mutation getMutation() { return mutation; diff --git a/src/java/org/apache/cassandra/db/IMutation.java b/src/java/org/apache/cassandra/db/IMutation.java index 44df104c20..3e037c30df 100644 --- a/src/java/org/apache/cassandra/db/IMutation.java +++ b/src/java/org/apache/cassandra/db/IMutation.java @@ -30,4 +30,15 @@ public interface IMutation public String toString(boolean shallow); public void addAll(IMutation m); public Collection getColumnFamilies(); + + /** + * Call to increment underlying network buffer refcount + * So we can avoid recycling too soon + */ + public void retain(); + + /** + * Call to decrement underlying network buffer refcount + */ + public void release(); } diff --git a/src/java/org/apache/cassandra/db/Mutation.java b/src/java/org/apache/cassandra/db/Mutation.java index b64c67583c..6eb56b7a08 100644 --- a/src/java/org/apache/cassandra/db/Mutation.java +++ b/src/java/org/apache/cassandra/db/Mutation.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.*; +import org.apache.cassandra.transport.Frame; import org.apache.commons.lang3.StringUtils; import org.apache.cassandra.config.CFMetaData; @@ -34,12 +35,15 @@ import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.net.MessageOut; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.utils.ByteBufferUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; // TODO convert this to a Builder pattern instead of encouraging M.add directly, // which is less-efficient since we have to keep a mutable HashMap around public class Mutation implements IMutation { public static final MutationSerializer serializer = new MutationSerializer(); + private static final Logger logger = LoggerFactory.getLogger(Mutation.class); public static final String FORWARD_TO = "FWD_TO"; public static final String FORWARD_FROM = "FWD_FRM"; @@ -52,6 +56,8 @@ public class Mutation implements IMutation // map of column family id to mutations for that column family. private final Map modifications; + private Frame sourceFrame; + public Mutation(String keyspaceName, ByteBuffer key) { this(keyspaceName, key, new HashMap()); @@ -81,7 +87,10 @@ public class Mutation implements IMutation public Mutation copy() { - return new Mutation(keyspaceName, key, new HashMap<>(modifications)); + Mutation copy = new Mutation(keyspaceName, key, new HashMap<>(modifications)); + copy.setSourceFrame(getSourceFrame()); + + return copy; } public String getKeyspaceName() @@ -104,6 +113,20 @@ public class Mutation implements IMutation return modifications.values(); } + @Override + public void retain() + { + if (sourceFrame != null) + sourceFrame.retain(); + } + + @Override + public void release() + { + if (sourceFrame != null) + sourceFrame.release(); + } + public ColumnFamily getColumnFamily(UUID cfId) { return modifications.get(cfId); @@ -206,6 +229,8 @@ public class Mutation implements IMutation */ public void apply() { + assert sourceFrame == null || sourceFrame.body.refCnt() > 0; + Keyspace ks = Keyspace.open(keyspaceName); ks.apply(this, ks.metadata.durableWrites); } @@ -265,6 +290,16 @@ public class Mutation implements IMutation return mutation; } + public Frame getSourceFrame() + { + return sourceFrame; + } + + public void setSourceFrame(Frame sourceFrame) + { + this.sourceFrame = sourceFrame; + } + public static class MutationSerializer implements IVersionedSerializer { public void serialize(Mutation mutation, DataOutputPlus out, int version) throws IOException diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index c0f2f26c4b..8e892e7200 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -349,10 +349,19 @@ public final class MessagingService implements MessagingServiceMBean }); } - if (expiredCallbackInfo.shouldHint()) + Mutation mutation = (Mutation) ((WriteCallbackInfo) expiredCallbackInfo).sentMessage.payload; + + try { - Mutation mutation = (Mutation) ((WriteCallbackInfo) expiredCallbackInfo).sentMessage.payload; - return StorageProxy.submitHint(mutation, expiredCallbackInfo.target, null); + if (expiredCallbackInfo.shouldHint()) + { + return StorageProxy.submitHint(mutation, expiredCallbackInfo.target, null); + } + } + finally + { + //We serialized a hint so we don't need this mutation anymore + mutation.release(); } return null; @@ -570,6 +579,11 @@ public final class MessagingService implements MessagingServiceMBean { assert message.verb == Verb.MUTATION || message.verb == Verb.COUNTER_MUTATION; int messageId = nextId(); + + //keep the underlying buffer around till the request completes or times out and + //a hint is stored + message.payload.retain(); + CallbackInfo previous = callbacks.put(messageId, new WriteCallbackInfo(to, cb, diff --git a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java b/src/java/org/apache/cassandra/net/ResponseVerbHandler.java index 1d9aa98ef6..1e1a278c5f 100644 --- a/src/java/org/apache/cassandra/net/ResponseVerbHandler.java +++ b/src/java/org/apache/cassandra/net/ResponseVerbHandler.java @@ -19,6 +19,7 @@ package org.apache.cassandra.net; import java.util.concurrent.TimeUnit; +import org.apache.cassandra.db.IMutation; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -52,5 +53,11 @@ public class ResponseVerbHandler implements IVerbHandler MessagingService.instance().maybeAddLatency(cb, message.from, latency); cb.response(message); } + + // We don't need to track the mutation anymore since write succeeded + if (callbackInfo instanceof WriteCallbackInfo) + { + ((IMutation)((WriteCallbackInfo) callbackInfo).sentMessage.payload).release(); + } } } diff --git a/src/java/org/apache/cassandra/service/QueryState.java b/src/java/org/apache/cassandra/service/QueryState.java index 12fc3926c2..f2e0809d8a 100644 --- a/src/java/org/apache/cassandra/service/QueryState.java +++ b/src/java/org/apache/cassandra/service/QueryState.java @@ -20,6 +20,7 @@ package org.apache.cassandra.service; import java.util.UUID; import org.apache.cassandra.tracing.Tracing; +import org.apache.cassandra.transport.Frame; import org.apache.cassandra.utils.FBUtilities; /** @@ -30,6 +31,7 @@ public class QueryState private final ClientState clientState; private volatile long clock; private volatile UUID preparedTracingSession; + private Frame sourceFrame; public QueryState(ClientState clientState) { @@ -60,6 +62,16 @@ public class QueryState return clock; } + public Frame getSourceFrame() + { + return sourceFrame; + } + + public void setSourceFrame(Frame sourceFrame) + { + this.sourceFrame = sourceFrame; + } + public boolean traceNextQuery() { if (preparedTracingSession != null) diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 2cbc475f1a..890315abc8 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -451,12 +451,14 @@ public class StorageProxy implements StorageProxyMBean final String localDataCenter = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddress()); long startTime = System.nanoTime(); - List responseHandlers = new ArrayList(mutations.size()); + List responseHandlers = new ArrayList<>(mutations.size()); try { for (IMutation mutation : mutations) { + mutation.retain(); + if (mutation instanceof CounterMutation) { responseHandlers.add(mutateCounter((CounterMutation)mutation, localDataCenter)); @@ -517,6 +519,13 @@ public class StorageProxy implements StorageProxyMBean } finally { + //Release the mutations we dispatched so far. + //An exception may be thrown at anytime. + //We can infer the mutations that were dispatched from this list + Iterator it = mutations.iterator(); + for (int i = 0; i < responseHandlers.size(); i++) + it.next().release(); + writeMetrics.addNano(System.nanoTime() - startTime); } } @@ -781,72 +790,79 @@ public class StorageProxy implements StorageProxyMBean MessageOut message = null; boolean insertLocal = false; - for (InetAddress destination : targets) + + + mutation.retain(); + try { - // avoid OOMing due to excess hints. we need to do this check even for "live" nodes, since we can - // still generate hints for those if it's overloaded or simply dead but not yet known-to-be-dead. - // The idea is that if we have over maxHintsInProgress hints in flight, this is probably due to - // a small number of nodes causing problems, so we should avoid shutting down writes completely to - // healthy nodes. Any node with no hintsInProgress is considered healthy. - if (StorageMetrics.totalHintsInProgress.count() > maxHintsInProgress - && (getHintsInProgressFor(destination).get() > 0 && shouldHint(destination))) + for (InetAddress destination : targets) { - throw new OverloadedException("Too many in flight hints: " + StorageMetrics.totalHintsInProgress.count()); - } - - if (FailureDetector.instance.isAlive(destination)) - { - if (destination.equals(FBUtilities.getBroadcastAddress()) && OPTIMIZE_LOCAL_REQUESTS) + // avoid OOMing due to excess hints. we need to do this check even for "live" nodes, since we can + // still generate hints for those if it's overloaded or simply dead but not yet known-to-be-dead. + // The idea is that if we have over maxHintsInProgress hints in flight, this is probably due to + // a small number of nodes causing problems, so we should avoid shutting down writes completely to + // healthy nodes. Any node with no hintsInProgress is considered healthy. + if (StorageMetrics.totalHintsInProgress.count() > maxHintsInProgress + && (getHintsInProgressFor(destination).get() > 0 && shouldHint(destination))) { - insertLocal = true; + throw new OverloadedException("Too many in flight hints: " + StorageMetrics.totalHintsInProgress.count()); } - else + + if (FailureDetector.instance.isAlive(destination)) { - // belongs on a different server - if (message == null) - message = mutation.createMessage(); - String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(destination); - // direct writes to local DC or old Cassandra versions - // (1.1 knows how to forward old-style String message IDs; updated to int in 2.0) - if (localDataCenter.equals(dc)) + if (destination.equals(FBUtilities.getBroadcastAddress()) && OPTIMIZE_LOCAL_REQUESTS) { - MessagingService.instance().sendRR(message, destination, responseHandler, true); - } - else + insertLocal = true; + } else { - Collection messages = (dcGroups != null) ? dcGroups.get(dc) : null; - if (messages == null) + // belongs on a different server + if (message == null) + message = mutation.createMessage(); + String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(destination); + // direct writes to local DC or old Cassandra versions + // (1.1 knows how to forward old-style String message IDs; updated to int in 2.0) + if (localDataCenter.equals(dc)) { - messages = new ArrayList(3); // most DCs will have <= 3 replicas - if (dcGroups == null) - dcGroups = new HashMap>(); - dcGroups.put(dc, messages); + MessagingService.instance().sendRR(message, destination, responseHandler, true); + } else + { + Collection messages = (dcGroups != null) ? dcGroups.get(dc) : null; + if (messages == null) + { + messages = new ArrayList(3); // most DCs will have <= 3 replicas + if (dcGroups == null) + dcGroups = new HashMap>(); + dcGroups.put(dc, messages); + } + messages.add(destination); } - messages.add(destination); } + } else + { + if (!shouldHint(destination)) + continue; + + // Schedule a local hint + submitHint(mutation, destination, responseHandler); } } - else - { - if (!shouldHint(destination)) - continue; - // Schedule a local hint - submitHint(mutation, destination, responseHandler); + if (insertLocal) + insertLocal(mutation, responseHandler); + + if (dcGroups != null) + { + // for each datacenter, send the message to one node to relay the write to other replicas + if (message == null) + message = mutation.createMessage(); + + for (Collection dcTargets : dcGroups.values()) + sendMessagesToNonlocalDC(message, dcTargets, responseHandler); } } - - if (insertLocal) - insertLocal(mutation, responseHandler); - - if (dcGroups != null) + finally { - // for each datacenter, send the message to one node to relay the write to other replicas - if (message == null) - message = mutation.createMessage(); - - for (Collection dcTargets : dcGroups.values()) - sendMessagesToNonlocalDC(message, dcTargets, responseHandler); + mutation.release(); } } @@ -868,23 +884,30 @@ public class StorageProxy implements StorageProxyMBean { // local write that time out should be handled by LocalMutationRunnable assert !target.equals(FBUtilities.getBroadcastAddress()) : target; + mutation.retain(); HintRunnable runnable = new HintRunnable(target) { public void runMayThrow() { - int ttl = HintedHandOffManager.calculateHintTTL(mutation); - if (ttl > 0) + try { - logger.debug("Adding hint for {}", target); - writeHintForMutation(mutation, System.currentTimeMillis(), ttl, target); - // Notify the handler only for CL == ANY - if (responseHandler != null && responseHandler.consistencyLevel == ConsistencyLevel.ANY) - responseHandler.response(null); + int ttl = HintedHandOffManager.calculateHintTTL(mutation); + if (ttl > 0) + { + logger.debug("Adding hint for {}", target); + writeHintForMutation(mutation, System.currentTimeMillis(), ttl, target); + // Notify the handler only for CL == ANY + if (responseHandler != null && responseHandler.consistencyLevel == ConsistencyLevel.ANY) + responseHandler.response(null); + } else + { + logger.debug("Skipped writing hint for {} (ttl {})", target, ttl); + } } - else + finally { - logger.debug("Skipped writing hint for {} (ttl {})", target, ttl); + mutation.release(); } } }; @@ -948,15 +971,24 @@ public class StorageProxy implements StorageProxyMBean private static void insertLocal(final Mutation mutation, final AbstractWriteResponseHandler responseHandler) { + mutation.retain(); + StageManager.getStage(Stage.MUTATION).maybeExecuteImmediately(new LocalMutationRunnable() { public void runMayThrow() { - IMutation processed = SinkManager.processWriteRequest(mutation); - if (processed != null) + try { - ((Mutation) processed).apply(); - responseHandler.response(null); + IMutation processed = SinkManager.processWriteRequest(mutation); + if (processed != null) + { + ((Mutation) processed).apply(); + responseHandler.response(null); + } + } + finally + { + mutation.release(); } } }); @@ -1062,8 +1094,11 @@ public class StorageProxy implements StorageProxyMBean final AbstractWriteResponseHandler responseHandler, final String localDataCenter) { + mutation.retain(); + return new DroppableRunnable(MessagingService.Verb.COUNTER_MUTATION) { + @Override public void runMayThrow() throws OverloadedException, WriteTimeoutException { IMutation processed = SinkManager.processWriteRequest(mutation); @@ -1077,10 +1112,16 @@ public class StorageProxy implements StorageProxyMBean responseHandler.response(null); Set remotes = Sets.difference(ImmutableSet.copyOf(targets), - ImmutableSet.of(FBUtilities.getBroadcastAddress())); + ImmutableSet.of(FBUtilities.getBroadcastAddress())); if (!remotes.isEmpty()) sendToHintedEndpoints(result, remotes, responseHandler, localDataCenter); } + + @Override + public void cleanup() + { + mutation.release(); + } }; } @@ -2000,22 +2041,32 @@ public class StorageProxy implements StorageProxyMBean public final void run() { - if (TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - constructionTime) > DatabaseDescriptor.getTimeout(verb)) - { - MessagingService.instance().incrementDroppedMessages(verb); - return; - } - try { - runMayThrow(); + if (TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - constructionTime) > DatabaseDescriptor.getTimeout(verb)) + { + MessagingService.instance().incrementDroppedMessages(verb); + return; + } + + try + { + runMayThrow(); + } catch (Exception e) + { + throw new RuntimeException(e); + } } - catch (Exception e) + finally { - throw new RuntimeException(e); + cleanup(); } } + public void cleanup() + { + } + abstract protected void runMayThrow() throws Exception; } diff --git a/src/java/org/apache/cassandra/transport/CBUtil.java b/src/java/org/apache/cassandra/transport/CBUtil.java index eac687d44e..6cc6d47e11 100644 --- a/src/java/org/apache/cassandra/transport/CBUtil.java +++ b/src/java/org/apache/cassandra/transport/CBUtil.java @@ -29,9 +29,7 @@ import java.util.List; import java.util.Map; import java.util.UUID; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.ByteBufAllocator; -import io.netty.buffer.PooledByteBufAllocator; +import io.netty.buffer.*; import io.netty.util.CharsetUtil; import org.apache.cassandra.db.ConsistencyLevel; @@ -49,7 +47,6 @@ import org.apache.cassandra.utils.UUIDGen; public abstract class CBUtil { public static final ByteBufAllocator allocator = new PooledByteBufAllocator(true); - public static final ByteBufAllocator onHeapAllocator = new PooledByteBufAllocator(false); private CBUtil() {} @@ -300,7 +297,11 @@ public abstract class CBUtil if (length < 0) return null; ByteBuf slice = cb.readSlice(length); - return ByteBuffer.wrap(readRawBytes(slice)); + if (slice.nioBufferCount() == 1) + return slice.nioBuffer(); + else + return ByteBuffer.wrap(readRawBytes(slice)); + } public static void writeValue(byte[] bytes, ByteBuf cb) diff --git a/src/java/org/apache/cassandra/transport/Frame.java b/src/java/org/apache/cassandra/transport/Frame.java index 3e66ff76ce..b29e80c1e5 100644 --- a/src/java/org/apache/cassandra/transport/Frame.java +++ b/src/java/org/apache/cassandra/transport/Frame.java @@ -1,3 +1,4 @@ + /* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file @@ -65,9 +66,14 @@ public class Frame this.body = body; } - public void release() + public void retain() { - body.release(); + body.retain(); + } + + public boolean release() + { + return body.release(); } public static Frame create(Message.Type type, int streamId, int version, EnumSet flags, ByteBuf body) @@ -226,7 +232,9 @@ public class Frame return; // extract body - ByteBuf body = CBUtil.allocator.buffer((int) bodyLength).writeBytes(buffer.duplicate().slice(idx, (int) bodyLength)); + ByteBuf body = buffer.slice(idx, (int) bodyLength); + body.retain(); + idx += bodyLength; buffer.readerIndex(idx); diff --git a/src/java/org/apache/cassandra/transport/FrameCompressor.java b/src/java/org/apache/cassandra/transport/FrameCompressor.java index 23bfce7295..c2fc662735 100644 --- a/src/java/org/apache/cassandra/transport/FrameCompressor.java +++ b/src/java/org/apache/cassandra/transport/FrameCompressor.java @@ -73,7 +73,7 @@ public interface FrameCompressor public Frame compress(Frame frame) throws IOException { byte[] input = CBUtil.readRawBytes(frame.body); - ByteBuf output = CBUtil.onHeapAllocator.buffer(Snappy.maxCompressedLength(input.length)); + ByteBuf output = CBUtil.allocator.heapBuffer(Snappy.maxCompressedLength(input.length)); try { @@ -101,7 +101,7 @@ public interface FrameCompressor if (!Snappy.isValidCompressedBuffer(input, 0, input.length)) throw new ProtocolException("Provided frame does not appear to be Snappy compressed"); - ByteBuf output = CBUtil.onHeapAllocator.buffer(Snappy.uncompressedLength(input)); + ByteBuf output = CBUtil.allocator.heapBuffer(Snappy.uncompressedLength(input)); try { @@ -151,7 +151,7 @@ public interface FrameCompressor byte[] input = CBUtil.readRawBytes(frame.body); int maxCompressedLength = compressor.maxCompressedLength(input.length); - ByteBuf outputBuf = CBUtil.onHeapAllocator.buffer(INTEGER_BYTES + maxCompressedLength); + ByteBuf outputBuf = CBUtil.allocator.heapBuffer(INTEGER_BYTES + maxCompressedLength); byte[] output = outputBuf.array(); int outputOffset = outputBuf.arrayOffset(); @@ -189,7 +189,7 @@ public interface FrameCompressor | ((input[2] & 0xFF) << 8) | ((input[3] & 0xFF)); - ByteBuf output = CBUtil.onHeapAllocator.buffer(uncompressedLength); + ByteBuf output = CBUtil.allocator.heapBuffer(uncompressedLength); try { diff --git a/src/java/org/apache/cassandra/transport/Message.java b/src/java/org/apache/cassandra/transport/Message.java index 9e8719eb92..f27d5452e6 100644 --- a/src/java/org/apache/cassandra/transport/Message.java +++ b/src/java/org/apache/cassandra/transport/Message.java @@ -124,9 +124,9 @@ public abstract class Message } public final Type type; - protected volatile Connection connection; - private volatile int streamId; - private volatile Frame sourceFrame; + protected Connection connection; + private int streamId; + private Frame sourceFrame = null; protected Message(Type type) { @@ -360,10 +360,8 @@ public abstract class Message for (ChannelHandlerContext channel : channels) channel.flush(); for (FlushItem item : flushed) - { - if (item.response.getSourceFrame().body.refCnt() > 0) - item.response.getSourceFrame().release(); - } + item.response.getSourceFrame().release(); + channels.clear(); flushed.clear(); runsSinceFlush = 0; @@ -407,6 +405,7 @@ public abstract class Message assert request.connection() instanceof ServerConnection; connection = (ServerConnection)request.connection(); QueryState qstate = connection.validateNewMessage(request.type, connection.getVersion(), request.getStreamId()); + qstate.setSourceFrame(request.getSourceFrame()); logger.debug("Received: {}, v={}", request, connection.getVersion()); diff --git a/src/java/org/apache/cassandra/utils/ExpiringMap.java b/src/java/org/apache/cassandra/utils/ExpiringMap.java index 7eec40e0ce..e7b626cd5e 100644 --- a/src/java/org/apache/cassandra/utils/ExpiringMap.java +++ b/src/java/org/apache/cassandra/utils/ExpiringMap.java @@ -91,10 +91,12 @@ public class ExpiringMap { if (entry.getValue().isReadyToDieAt(start)) { - cache.remove(entry.getKey()); - n++; - if (postExpireHook != null) - postExpireHook.apply(Pair.create(entry.getKey(), entry.getValue())); + if (cache.remove(entry.getKey()) != null) + { + n++; + if (postExpireHook != null) + postExpireHook.apply(Pair.create(entry.getKey(), entry.getValue())); + } } } logger.trace("Expired {} entries", n);