api, core: support zero copy into protobuf

api/src/main/java/io/grpc/Detachable.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2021 The gRPC Authors
+ *
+ * Licensed 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 io.grpc;
+
+/**
+ * A <i>Detachable</i> encapsulates an object that can be forked with underlying resources
+ * detached and transferred to a new instance. The forked instance takes over the ownership
+ * of resources and is responsible for releasing after use. The forked instance preserves
+ * states of detached resources. Resources can be consumed through the forked instance as if
+ * being continually consumed through the original instance. The original instance discards
+ * states of detached resources and is no longer consumable as if the resources are exhausted.
+ */
+@ExperimentalApi("TODO")
+public interface Detachable {
+
+  /**
+   * Fork a new instance with underlying resources detached from this instance and
+   * transferred to the new instance.
+   *
+   * @throws IllegalStateException if the underlying resources have already been detached.
+   */
+  public Detachable detach();
+}

api/src/main/java/io/grpc/HasByteBuffer.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2020 The gRPC Authors
+ *
+ * Licensed 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 io.grpc;
+
+import java.nio.ByteBuffer;
+import javax.annotation.Nullable;
+
+/**
+ * Extension to an {@link java.io.InputStream} whose content can be accessed as {@link
+ * ByteBuffer}s.
+ *
+ * <p>This can be used for optimizing the case for the consumer of a {@link ByteBuffer}-backed
+ * input stream supports efficient reading from {@link ByteBuffer}s directly. This turns the reader
+ * interface from an {@link java.io.InputStream} to {@link ByteBuffer}s, without copying the
+ * content to a byte array and read from it.
+ */
+@ExperimentalApi("https://github.com/grpc/grpc-java/issues/7387")
+public interface HasByteBuffer {
+
+  /**
+   * Indicates whether or not {@link #getByteBuffer} operation is supported.
+   */
+  boolean byteBufferSupported();
+
+  /**
+   * Gets a {@link ByteBuffer} containing some bytes of the content next to be read, or {@code
+   * null} if has reached end of the content. The number of bytes contained in the returned buffer
+   * is implementation specific. Calling this method does not change the position of the input
+   * stream. The returned buffer's content should not be modified, but the position, limit, and
+   * mark may be changed. Operations for changing the position, limit, and mark of the returned
+   * buffer does not affect the position, limit, and mark of this input stream. This is an optional
+   * method, so callers should first check {@link #byteBufferSupported}.
+   *
+   * @throws UnsupportedOperationException if this operation is not supported.
+   */
+  @Nullable
+  ByteBuffer getByteBuffer();
+}

core/src/main/java/io/grpc/internal/AbstractReadableBuffer.java

@@ -16,6 +16,8 @@
 
 package io.grpc.internal;
 
+import java.nio.ByteBuffer;
+
 /**
  * Abstract base class for {@link ReadableBuffer} implementations.
  */
@@ -45,6 +47,29 @@ public int arrayOffset() {
     throw new UnsupportedOperationException();
   }
 
+  @Override
+  public boolean markSupported() {
+    return false;
+  }
+
+  @Override
+  public void mark() {}
+
+  @Override
+  public void reset() {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public boolean byteBufferSupported() {
+    return false;
+  }
+
+  @Override
+  public ByteBuffer getByteBuffer() {
+    throw new UnsupportedOperationException();
+  }
+
   @Override
   public void close() {}
 

core/src/main/java/io/grpc/internal/CompositeReadableBuffer.java

@@ -20,8 +20,10 @@
 import java.io.OutputStream;
 import java.nio.Buffer;
 import java.nio.ByteBuffer;
+import java.nio.InvalidMarkException;
 import java.util.ArrayDeque;
-import java.util.Queue;
+import java.util.Deque;
+import javax.annotation.Nullable;
 
 /**
  * A {@link ReadableBuffer} that is composed of 0 or more {@link ReadableBuffer}s. This provides a
@@ -33,15 +35,17 @@
  */
 public class CompositeReadableBuffer extends AbstractReadableBuffer {
 
+  private final Deque<ReadableBuffer> readableBuffers;
+  private Deque<ReadableBuffer> rewindableBuffers;
   private int readableBytes;
-  private final Queue<ReadableBuffer> buffers;
+  private boolean marked;
 
   public CompositeReadableBuffer(int initialCapacity) {
-    buffers = new ArrayDeque<>(initialCapacity);
+    readableBuffers = new ArrayDeque<>(initialCapacity);
   }
 
   public CompositeReadableBuffer() {
-    buffers = new ArrayDeque<>();
+    readableBuffers = new ArrayDeque<>();
   }
 
   /**
@@ -51,16 +55,24 @@ public CompositeReadableBuffer() {
    * this {@code CompositeBuffer}.
    */
   public void addBuffer(ReadableBuffer buffer) {
+    boolean markHead = marked && readableBuffers.isEmpty();
+    enqueueBuffer(buffer);
+    if (markHead) {
+      readableBuffers.peek().mark();
+    }
+  }
+
+  private void enqueueBuffer(ReadableBuffer buffer) {
     if (!(buffer instanceof CompositeReadableBuffer)) {
-      buffers.add(buffer);
+      readableBuffers.add(buffer);
       readableBytes += buffer.readableBytes();
       return;
     }
 
     CompositeReadableBuffer compositeBuffer = (CompositeReadableBuffer) buffer;
-    while (!compositeBuffer.buffers.isEmpty()) {
-      ReadableBuffer subBuffer = compositeBuffer.buffers.remove();
-      buffers.add(subBuffer);
+    while (!compositeBuffer.readableBuffers.isEmpty()) {
+      ReadableBuffer subBuffer = compositeBuffer.readableBuffers.remove();
+      readableBuffers.add(subBuffer);
     }
     readableBytes += compositeBuffer.readableBytes;
     compositeBuffer.readableBytes = 0;
@@ -158,22 +170,27 @@ public ReadableBuffer readBytes(int length) {
     ReadableBuffer newBuffer = null;
     CompositeReadableBuffer newComposite = null;
     do {
-      ReadableBuffer buffer = buffers.peek();
+      ReadableBuffer buffer = readableBuffers.peek();
       int readable = buffer.readableBytes();
       ReadableBuffer readBuffer;
       if (readable > length) {
         readBuffer = buffer.readBytes(length);
         length = 0;
       } else {
-        readBuffer = buffers.poll();
+        if (marked) {
+          readBuffer = buffer.readBytes(readable);
+          advanceBuffer();
+        } else {
+          readBuffer = readableBuffers.poll();
+        }
         length -= readable;
       }
       if (newBuffer == null) {
         newBuffer = readBuffer;
       } else {
         if (newComposite == null) {
-          newComposite =
-              new CompositeReadableBuffer(length == 0 ? 2 : Math.min(buffers.size() + 2, 16));
+          newComposite = new CompositeReadableBuffer(
+              length == 0 ? 2 : Math.min(readableBuffers.size() + 2, 16));
           newComposite.addBuffer(newBuffer);
           newBuffer = newComposite;
         }
@@ -183,10 +200,77 @@ public ReadableBuffer readBytes(int length) {
     return newBuffer;
   }
 
+  @Override
+  public boolean markSupported() {
+    for (ReadableBuffer buffer : readableBuffers) {
+      if (!buffer.markSupported()) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  @Override
+  public void mark() {
+    if (rewindableBuffers == null) {
+      rewindableBuffers = new ArrayDeque<>(Math.min(readableBuffers.size(), 16));
+    }
+    while (!rewindableBuffers.isEmpty()) {
+      rewindableBuffers.remove().close();
+    }
+    marked = true;
+    ReadableBuffer buffer = readableBuffers.peek();
+    if (buffer != null) {
+      buffer.mark();
+    }
+  }
+
+  @Override
+  public void reset() {
+    if (!marked) {
+      throw new InvalidMarkException();
+    }
+    ReadableBuffer buffer;
+    if ((buffer = readableBuffers.peek()) != null) {
+      int currentRemain = buffer.readableBytes();
+      buffer.reset();
+      readableBytes += (buffer.readableBytes() - currentRemain);
+    }
+    while ((buffer = rewindableBuffers.pollLast()) != null) {
+      buffer.reset();
+      readableBuffers.addFirst(buffer);
+      readableBytes += buffer.readableBytes();
+    }
+  }
+
+  @Override
+  public boolean byteBufferSupported() {
+    for (ReadableBuffer buffer : readableBuffers) {
+      if (!buffer.byteBufferSupported()) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  @Nullable
+  @Override
+  public ByteBuffer getByteBuffer() {
+    if (readableBuffers.isEmpty()) {
+      return null;
+    }
+    return readableBuffers.peek().getByteBuffer();
+  }
+
   @Override
   public void close() {
-    while (!buffers.isEmpty()) {
-      buffers.remove().close();
+    while (!readableBuffers.isEmpty()) {
+      readableBuffers.remove().close();
+    }
+    if (rewindableBuffers != null) {
+      while (!rewindableBuffers.isEmpty()) {
+        rewindableBuffers.remove().close();
+      }
     }
   }
 
@@ -197,12 +281,12 @@ public void close() {
   private <T> int execute(ReadOperation<T> op, int length, T dest, int value) throws IOException {
     checkReadable(length);
 
-    if (!buffers.isEmpty()) {
+    if (!readableBuffers.isEmpty()) {
       advanceBufferIfNecessary();
     }
 
-    for (; length > 0 && !buffers.isEmpty(); advanceBufferIfNecessary()) {
-      ReadableBuffer buffer = buffers.peek();
+    for (; length > 0 && !readableBuffers.isEmpty(); advanceBufferIfNecessary()) {
+      ReadableBuffer buffer = readableBuffers.peek();
       int lengthToCopy = Math.min(length, buffer.readableBytes());
 
       // Perform the read operation for this buffer.
@@ -232,9 +316,24 @@ private <T> int executeNoThrow(NoThrowReadOperation<T> op, int length, T dest, i
    * If the current buffer is exhausted, removes and closes it.
    */
   private void advanceBufferIfNecessary() {
-    ReadableBuffer buffer = buffers.peek();
+    ReadableBuffer buffer = readableBuffers.peek();
     if (buffer.readableBytes() == 0) {
-      buffers.remove().close();
+      advanceBuffer();
+    }
+  }
+
+  /**
+   * Removes one buffer from the front and closes it.
+   */
+  private void advanceBuffer() {
+    if (marked) {
+      rewindableBuffers.add(readableBuffers.remove());
+      ReadableBuffer next = readableBuffers.peek();
+      if (next != null) {
+        next.mark();
+      }
+    } else {
+      readableBuffers.remove().close();
     }
   }
 

core/src/main/java/io/grpc/internal/ForwardingReadableBuffer.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
+import javax.annotation.Nullable;
 
 /**
  * Base class for a wrapper around another {@link ReadableBuffer}.
@@ -96,6 +97,32 @@ public int arrayOffset() {
     return buf.arrayOffset();
   }
 
+  @Override
+  public boolean markSupported() {
+    return buf.markSupported();
+  }
+
+  @Override
+  public void mark() {
+    buf.mark();
+  }
+
+  @Override
+  public void reset() {
+    buf.reset();
+  }
+
+  @Override
+  public boolean byteBufferSupported() {
+    return buf.byteBufferSupported();
+  }
+
+  @Nullable
+  @Override
+  public ByteBuffer getByteBuffer() {
+    return buf.getByteBuffer();
+  }
+
   @Override
   public void close() {
     buf.close();