From ff27eb304c84a5b542b8a5da1712f289fa45ba81 Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Mon, 14 Sep 2015 19:47:06 +0100 Subject: [PATCH] Implement a more efficient skipBytes (CASSANDRA-10322) Avoids garbage production during a call to skipBytes on any DataInputPlus, and strengthens the skipBytes contract to always skip the requested number of bytes unless EOF is encountered. Also moves FileUtils.skipBytesFully to DataInputPlus. patch by benedict; reviewed by jbellis for CASSANDRA-10322 --- .../org/apache/cassandra/db/DeletionTime.java | 2 +- .../apache/cassandra/db/RowIndexEntry.java | 4 ++-- .../cassandra/db/marshal/AbstractType.java | 2 +- .../cassandra/io/util/DataInputPlus.java | 21 +++++++++++++----- .../apache/cassandra/io/util/FileUtils.java | 12 ---------- .../io/util/RebufferingInputStream.java | 22 ++++++++++++------- .../org/apache/cassandra/net/MessageIn.java | 2 +- .../cassandra/utils/ByteBufferUtil.java | 6 ++--- 8 files changed, 38 insertions(+), 33 deletions(-) diff --git a/src/java/org/apache/cassandra/db/DeletionTime.java b/src/java/org/apache/cassandra/db/DeletionTime.java index 343a6c2ee1..919c603188 100644 --- a/src/java/org/apache/cassandra/db/DeletionTime.java +++ b/src/java/org/apache/cassandra/db/DeletionTime.java @@ -173,7 +173,7 @@ public class DeletionTime implements Comparable, IMeasurableMemory public void skip(DataInputPlus in) throws IOException { - FileUtils.skipBytesFully(in, 4 + 8); + in.skipBytesFully(4 + 8); } public long serializedSize(DeletionTime delTime) diff --git a/src/java/org/apache/cassandra/db/RowIndexEntry.java b/src/java/org/apache/cassandra/db/RowIndexEntry.java index 198e8908a9..4e2f063638 100644 --- a/src/java/org/apache/cassandra/db/RowIndexEntry.java +++ b/src/java/org/apache/cassandra/db/RowIndexEntry.java @@ -217,7 +217,7 @@ public class RowIndexEntry implements IMeasurableMemory for (int i = 0; i < entries; i++) columnsIndex.add(idxSerializer.deserialize(in)); - FileUtils.skipBytesFully(in, entries * TypeSizes.sizeof(0)); + in.skipBytesFully(entries * TypeSizes.sizeof(0)); return new IndexedEntry(position, deletionTime, headerLength, columnsIndex); } @@ -247,7 +247,7 @@ public class RowIndexEntry implements IMeasurableMemory if (size <= 0) return; - FileUtils.skipBytesFully(in, size); + in.skipBytesFully(size); } public int serializedSize(RowIndexEntry rie) diff --git a/src/java/org/apache/cassandra/db/marshal/AbstractType.java b/src/java/org/apache/cassandra/db/marshal/AbstractType.java index 5c46823889..a0d915f34c 100644 --- a/src/java/org/apache/cassandra/db/marshal/AbstractType.java +++ b/src/java/org/apache/cassandra/db/marshal/AbstractType.java @@ -395,7 +395,7 @@ public abstract class AbstractType implements Comparator { int length = valueLengthIfFixed(); if (length >= 0) - FileUtils.skipBytesFully(in, length); + in.skipBytesFully(length); else ByteBufferUtil.skipWithVIntLength(in); } diff --git a/src/java/org/apache/cassandra/io/util/DataInputPlus.java b/src/java/org/apache/cassandra/io/util/DataInputPlus.java index a0294279a5..7c29ee18a6 100644 --- a/src/java/org/apache/cassandra/io/util/DataInputPlus.java +++ b/src/java/org/apache/cassandra/io/util/DataInputPlus.java @@ -17,10 +17,7 @@ */ package org.apache.cassandra.io.util; -import java.io.DataInput; -import java.io.DataInputStream; -import java.io.IOException; -import java.io.InputStream; +import java.io.*; import org.apache.cassandra.utils.vint.VIntCoding; @@ -29,7 +26,6 @@ import org.apache.cassandra.utils.vint.VIntCoding; */ public interface DataInputPlus extends DataInput { - default long readVInt() throws IOException { return VIntCoding.readVInt(this); @@ -47,6 +43,21 @@ public interface DataInputPlus extends DataInput return VIntCoding.readUnsignedVInt(this); } + /** + * Always skips the requested number of bytes, unless EOF is reached + * + * @param n number of bytes to skip + * @return number of bytes skipped + */ + public int skipBytes(int n) throws IOException; + + public default void skipBytesFully(int n) throws IOException + { + int skipped = skipBytes(n); + if (skipped != n) + throw new EOFException("EOF after " + skipped + " bytes out of " + n); + } + /** * Wrapper around an InputStream that provides no buffering but can decode varints */ diff --git a/src/java/org/apache/cassandra/io/util/FileUtils.java b/src/java/org/apache/cassandra/io/util/FileUtils.java index 8b7b1e12f4..78eeb8f1f9 100644 --- a/src/java/org/apache/cassandra/io/util/FileUtils.java +++ b/src/java/org/apache/cassandra/io/util/FileUtils.java @@ -458,18 +458,6 @@ public class FileUtils dir.deleteOnExit(); } - public static void skipBytesFully(DataInput in, int bytes) throws IOException - { - int n = 0; - while (n < bytes) - { - int skipped = in.skipBytes(bytes - n); - if (skipped == 0) - throw new EOFException("EOF after " + n + " bytes out of " + bytes); - n += skipped; - } - } - public static void handleCorruptSSTable(CorruptSSTableException e) { if (!StorageService.instance.isSetupCompleted()) diff --git a/src/java/org/apache/cassandra/io/util/RebufferingInputStream.java b/src/java/org/apache/cassandra/io/util/RebufferingInputStream.java index 958a8157d6..306874628d 100644 --- a/src/java/org/apache/cassandra/io/util/RebufferingInputStream.java +++ b/src/java/org/apache/cassandra/io/util/RebufferingInputStream.java @@ -112,16 +112,22 @@ public abstract class RebufferingInputStream extends InputStream implements Data @Override public int skipBytes(int n) throws IOException { - int skipped = 0; - - while (skipped < n) + if (n < 0) + return 0; + int requested = n; + int position = buffer.position(), limit = buffer.limit(), remaining; + while ((remaining = limit - position) < n) { - int skippedThisTime = (int)skip(n - skipped); - if (skippedThisTime <= 0) break; - skipped += skippedThisTime; + n -= remaining; + buffer.position(limit); + reBuffer(); + position = buffer.position(); + limit = buffer.limit(); + if (position == limit) + return requested - n; } - - return skipped; + buffer.position(position + n); + return requested; } @Override diff --git a/src/java/org/apache/cassandra/net/MessageIn.java b/src/java/org/apache/cassandra/net/MessageIn.java index 82f400033c..64b8e81cab 100644 --- a/src/java/org/apache/cassandra/net/MessageIn.java +++ b/src/java/org/apache/cassandra/net/MessageIn.java @@ -88,7 +88,7 @@ public class MessageIn if (callback == null) { // reply for expired callback. we'll have to skip it. - FileUtils.skipBytesFully(in, payloadSize); + in.skipBytesFully(payloadSize); return null; } serializer = (IVersionedSerializer) callback.serializer; diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java index 70d4bd5240..27f46b60be 100644 --- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java +++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java @@ -357,7 +357,7 @@ public class ByteBufferUtil if (length < 0) throw new IOException("Corrupt (negative) value length encountered"); - FileUtils.skipBytesFully(in, length); + in.skipBytesFully(length); } /* @return An unsigned short in an integer. */ @@ -387,10 +387,10 @@ public class ByteBufferUtil * @return null * @throws IOException if an I/O error occurs. */ - public static void skipShortLength(DataInput in) throws IOException + public static void skipShortLength(DataInputPlus in) throws IOException { int skip = readShortLength(in); - FileUtils.skipBytesFully(in, skip); + in.skipBytesFully(skip); } public static ByteBuffer read(DataInput in, int length) throws IOException