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

src/java/org/apache/cassandra/db/DeletionTime.java

@@ -173,7 +173,7 @@ public class DeletionTime implements Comparable<DeletionTime>, IMeasurableMemory
 
         public void skip(DataInputPlus in) throws IOException
         {
-            FileUtils.skipBytesFully(in, 4 + 8);
+            in.skipBytesFully(4 + 8);
         }
 
         public long serializedSize(DeletionTime delTime)

src/java/org/apache/cassandra/db/RowIndexEntry.java

@@ -217,7 +217,7 @@ public class RowIndexEntry<T> 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<T> implements IMeasurableMemory
             if (size <= 0)
                 return;
 
-            FileUtils.skipBytesFully(in, size);
+            in.skipBytesFully(size);
         }
 
         public int serializedSize(RowIndexEntry<IndexHelper.IndexInfo> rie)

src/java/org/apache/cassandra/db/marshal/AbstractType.java

@@ -395,7 +395,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
     {
         int length = valueLengthIfFixed();
         if (length >= 0)
-            FileUtils.skipBytesFully(in, length);
+            in.skipBytesFully(length);
         else
             ByteBufferUtil.skipWithVIntLength(in);
     }

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
      */

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())

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

src/java/org/apache/cassandra/net/MessageIn.java

@@ -88,7 +88,7 @@ public class MessageIn<T>
             if (callback == null)
             {
                 // reply for expired callback.  we'll have to skip it.
-                FileUtils.skipBytesFully(in, payloadSize);
+                in.skipBytesFully(payloadSize);
                 return null;
             }
             serializer = (IVersionedSerializer<T2>) callback.serializer;

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