api, core: support zero copy into protobuf

[voidzcy]

A variant of #7330, taking the alternative codepath in protobuf.

Major changes of this PR:

• Added a HasByteBuffer API that allows the marshaller to access the backing ByteBuffers directly (cherry-picked from protobuf, api, core, netty: zero copy into protobuf #7330).
• Added a Detachable API that allows the application to take over the ownership of underlying buffers and close them later.

The new approach is to let the application implement a marshaller that passes ByteBuffers as immutable to protobuf:

• Wrap ByteBuffers to ByteString with UnsafeByteOperations
• Concatenate them as a RopeByteString
• Turn it to CodedInputStream

This requires buffers to be alive all the time through the proto messages used by the application (compared to the normal codepath that makes a copy of the bytes when parsed to proto messages and recycle buffers right after). The application is responsible for managing the lifetime of underlying buffers. An example marshaller that enables zero-copy codepath can be something like the following:

  class ZeroCopyMarshaller<T extends Message> implements PrototypeMarshaller<T> {
    private IdentityHashMap<T, InputStream> unclosedStreams = new IdentityHashMap<>();
    private final Parser<T> parser;

    ZeroCopyMarshaller(T defaultInstance) {
      parser = (Parser<T>) defaultInstance.getParserForType();
    }

    // ... ...

    @Override
    public T parse(InputStream stream) {
      CodedInputStream cis = null;
      boolean streamDetached = false;
      if (stream instanceof Detachable) {
        stream = ((Detachable) stream).detach();
        streamDetached = true;
      }
      try {
        if (stream instanceof KnownLength) {
          if (stream instanceof HasByteBuffer
              && ((HasByteBuffer) stream).byteBufferSupported()) {  // fastest path, zero copy
            List<ByteString> byteStrings = new ArrayList<>();
            stream.mark(stream.available());  // retain bytes in ByteBuffers so that calling skip() won't throw bytes away
            while (stream.available() != 0) {
              ByteBuffer buffer = ((HasByteBuffer) stream).getByteBuffer();
              byteStrings.add(UnsafeByteOperations.unsafeWrap(buffer));
              stream.skip(buffer.remaining());
            }
            stream.reset();
            cis = ByteString.copyFrom(byteStrings).newCodedInput();
          } else {
            // slightly slower path, copy into a byte array
            // see ProtoLiteUtils#MessageMarshaller
            // ...
          }
        }
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
      if (cis == null) {  // slowest path, via InputStream read
        cis = CodedInputStream.newInstance(stream);
      }

      // ... ...

      T message;
      try {
        message = parseFrom(cis);
      } catch (InvalidProtocolBufferException ipbe) {
        throw Status.INTERNAL.withDescription("Invalid protobuf byte sequence").withCause(ipbe).asRuntimeException();
      }
      if (streamDetached) {
        unclosedStreams.put(message, stream);
      }
      return message;
    }

    private T parseFrom(CodedInputStream stream) throws InvalidProtocolBufferException {
      T message = parser.parseFrom(stream);
      try {
        stream.checkLastTagWas(0);
        return message;
      } catch (InvalidProtocolBufferException e) {
        e.setUnfinishedMessage(message);
        throw e;
      }
    }

    // Application calls this method when no longer need the message.
    public void releaseMessage(T message) {
      InputStream stream = unclosedStreams.get(message);
      if (stream != null) {
        try {
          stream.close();
        } catch (IOException e) {
          throw new RuntimeException(e);
        }
        unclosedStreams.remove(message);
      }
    }
  }

[voidzcy]

@ejona86 Updated. PTAL and let me know if anything I am misunderstanding. Thanks you!

[ejona86]

The comments for Detachable are probably best discussed during API review and see what people think. I think this looks pretty good. Main thing is I'd hope is we don't need all the if (detached) checks in the implementation.

[voidzcy]

Updated descriptions in this PR and ExperimentalApi links in the code. Please let me know if any doc needs improvement. Thanks for the review!

[ejona86]

I think we need to "throw the user a bone" here and give them a better hint why this exists. Let's say something about how the detached instance can outlive the original instance. And how it may be useful in a Marshaller to perform delayed deserialization or when combined with HasByteBuffer.

An really, let's rephrase the documentation to describe this as "An extension of InputStream." I think it will make it much more clear what is going on.

[voidzcy]

Updated.

[veblush]

Thank you for the PR! My early benchmark shows that this can save ~20% of cpu time when used in the GCS benchmark client. (link)