Skip to content

[WIP] Zero-Copy Optimization for Triple Protocol #15700

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 2 commits into
base: 3.3
Choose a base branch
from
Draft

[WIP] Zero-Copy Optimization for Triple Protocol #15700

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2f9b8c5
Zero-Copy Optimization for Triple Protocol
finefuture Sep 21, 2025
eb56bc0
Enhance client.
finefuture Sep 24, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions dubbo-common/src/main/java/org/apache/dubbo/common/utils/TypeUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,4 +235,8 @@ static Class<?> getArgumentClass(Type type, int index) {
}
return null;
}

static Class<?> getSuperGenericType(Class<?> clazz, Class<?> interfaceClass, int index) {
return findActualTypeArgument(clazz, interfaceClass, index);
}
}
19 changes: 11 additions & 8 deletions ...servlet/src/main/java/org/apache/dubbo/rpc/protocol/tri/servlet/ServletStreamChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@

import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.remoting.http12.EmptyOutputStreamMessage;
import org.apache.dubbo.remoting.http12.HttpConstants;
import org.apache.dubbo.remoting.http12.HttpHeaderNames;
import org.apache.dubbo.remoting.http12.HttpHeaders;
Expand All @@ -38,6 +39,7 @@
import javax.servlet.http.HttpServletResponse;

import java.io.ByteArrayOutputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.HashMap;
Expand All @@ -48,7 +50,7 @@
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

final class ServletStreamChannel implements H2StreamChannel {
final class ServletStreamChannel implements H2StreamChannel<OutputStream> {

private static final Logger LOGGER = LoggerFactory.getLogger(ServletStreamChannel.class);

Expand Down Expand Up @@ -104,6 +106,7 @@ public void onWritePossible() {
}
}

@SuppressWarnings("unchecked")
private void flushQueue() {
if (writeQueue.isEmpty()) {
return;
Expand All @@ -114,7 +117,7 @@ private void flushQueue() {
if (obj instanceof HttpMetadata) {
writeHeaderInternal((HttpMetadata) obj);
} else if (obj instanceof HttpOutputMessage) {
writeMessageInternal((HttpOutputMessage) obj);
writeMessageInternal((HttpOutputMessage<OutputStream>) obj);
}
}
}
Expand Down Expand Up @@ -150,8 +153,8 @@ public CompletableFuture<Void> writeResetFrame(long errorCode) {
}

@Override
public Http2OutputMessage newOutputMessage(boolean endStream) {
return new Http2OutputMessageFrame(new ByteArrayOutputStream(256), endStream);
public Http2OutputMessage<OutputStream> newOutputMessage(boolean endStream) {
return new Http2OutputMessageFrame<>(new ByteArrayOutputStream(256), endStream);
}

@Override
Expand Down Expand Up @@ -213,7 +216,7 @@ private void writeHeaderInternal(HttpMetadata httpMetadata) {
}

@Override
public CompletableFuture<Void> writeMessage(HttpOutputMessage httpOutputMessage) {
public CompletableFuture<Void> writeMessage(HttpOutputMessage<OutputStream> httpOutputMessage) {
if (writeable.get()) {
flushQueue();
writeMessageInternal(httpOutputMessage);
Expand All @@ -223,11 +226,11 @@ public CompletableFuture<Void> writeMessage(HttpOutputMessage httpOutputMessage)
return completed();
}

private void writeMessageInternal(HttpOutputMessage httpOutputMessage) {
private void writeMessageInternal(HttpOutputMessage<OutputStream> httpOutputMessage) {
boolean endStream = false;
if (httpOutputMessage instanceof Http2OutputMessage) {
endStream = ((Http2OutputMessage) httpOutputMessage).isEndStream();
} else if (httpOutputMessage == HttpOutputMessage.EMPTY_MESSAGE) {
endStream = ((Http2OutputMessage<OutputStream>) httpOutputMessage).isEndStream();
} else if (httpOutputMessage == EmptyOutputStreamMessage.INSTANCE) {
endStream = true;
}
try {
Expand Down
33 changes: 24 additions & 9 deletions ...-triple-servlet/src/main/java/org/apache/dubbo/rpc/protocol/tri/servlet/TripleFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.remoting.http12.HttpVersion;
import org.apache.dubbo.remoting.http12.MessageTypeToken;
import org.apache.dubbo.remoting.http12.h1.Http1InputMessage;
import org.apache.dubbo.remoting.http12.h1.Http1ServerTransportListener;
import org.apache.dubbo.remoting.http12.h2.Http2InputMessageFrame;
Expand Down Expand Up @@ -56,6 +57,8 @@

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Arrays;
import java.util.Set;

Expand Down Expand Up @@ -101,8 +104,13 @@ private void handleHttp2(HttpServletRequest request, HttpServletResponse respons
AsyncContext context = request.startAsync(request, response);
ServletStreamChannel channel = new ServletStreamChannel(request, response, context);
try {
Http2TransportListener listener = determineHttp2ServerTransportListenerFactory(request.getContentType())
.newInstance(channel, ServletExchanger.getUrl(), FrameworkModel.defaultModel());
Http2TransportListener<InputStream, OutputStream> listener = determineHttp2ServerTransportListenerFactory(
request.getContentType())
.newInstance(
channel,
ServletExchanger.getUrl(),
FrameworkModel.defaultModel(),
new MessageTypeToken<InputStream, OutputStream>() {});

boolean isGrpc = listener instanceof GrpcHttp2ServerTransportListener;
channel.setGrpc(isGrpc);
Expand All @@ -123,15 +131,19 @@ private void handleHttp1(HttpServletRequest request, HttpServletResponse respons
AsyncContext context = request.startAsync(request, response);
ServletStreamChannel channel = new ServletStreamChannel(request, response, context);
try {
Http1ServerTransportListener listener = DefaultHttp11ServerTransportListenerFactory.INSTANCE.newInstance(
channel, ServletExchanger.getUrl(), FrameworkModel.defaultModel());
Http1ServerTransportListener<InputStream, OutputStream> listener =
DefaultHttp11ServerTransportListenerFactory.INSTANCE.newInstance(
channel,
ServletExchanger.getUrl(),
FrameworkModel.defaultModel(),
new MessageTypeToken<InputStream, OutputStream>() {});
channel.setGrpc(false);
context.setTimeout(resolveTimeout(request, false));
ServletInputStream is = request.getInputStream();
response.getOutputStream().setWriteListener(new TripleWriteListener(channel));

listener.onMetadata(new HttpMetadataAdapter(request));
listener.onData(new Http1InputMessage(
listener.onData(new Http1InputMessage<>(
is.available() == 0 ? StreamUtils.EMPTY : new ByteArrayInputStream(StreamUtils.readBytes(is))));
} catch (Throwable t) {
LOGGER.info("Failed to process request", t);
Expand Down Expand Up @@ -221,12 +233,15 @@ public void onStartAsync(AsyncEvent event) {}

private static final class TripleReadListener implements ReadListener {

private final Http2TransportListener listener;
private final Http2TransportListener<InputStream, OutputStream> listener;
private final ServletStreamChannel channel;
private final ServletInputStream input;
private final byte[] buffer = new byte[4 * 1024];

TripleReadListener(Http2TransportListener listener, ServletStreamChannel channel, ServletInputStream input) {
TripleReadListener(
Http2TransportListener<InputStream, OutputStream> listener,
ServletStreamChannel channel,
ServletInputStream input) {
this.listener = listener;
this.channel = channel;
this.input = input;
Expand All @@ -240,13 +255,13 @@ public void onDataAvailable() throws IOException {
return;
}
byte[] copy = Arrays.copyOf(buffer, length);
listener.onData(new Http2InputMessageFrame(new ByteArrayInputStream(copy), false));
listener.onData(new Http2InputMessageFrame<>(new ByteArrayInputStream(copy), false));
}
}

@Override
public void onAllDataRead() {
listener.onData(new Http2InputMessageFrame(StreamUtils.EMPTY, true));
listener.onData(new Http2InputMessageFrame<>(StreamUtils.EMPTY, true));
}

@Override
Expand Down
11 changes: 7 additions & 4 deletions ...src/main/java/org/apache/dubbo/rpc/protocol/tri/websocket/TripleBinaryMessageHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,20 +23,23 @@

import javax.websocket.MessageHandler;

import java.io.InputStream;
import java.io.OutputStream;
import java.nio.ByteBuffer;

public class TripleBinaryMessageHandler implements MessageHandler.Partial<ByteBuffer> {

private final WebSocketTransportListener webSocketTransportListener;
private final WebSocketTransportListener<InputStream, OutputStream> webSocketTransportListener;

public TripleBinaryMessageHandler(WebSocketTransportListener webSocketTransportListener) {
public TripleBinaryMessageHandler(
WebSocketTransportListener<InputStream, OutputStream> webSocketTransportListener) {
this.webSocketTransportListener = webSocketTransportListener;
}

@Override
public void onMessage(ByteBuffer messagePart, boolean last) {
Http2InputMessage http2InputMessage =
new Http2InputMessageFrame(new FinalFragmentByteArrayInputStream(messagePart.array(), last), false);
Http2InputMessage<InputStream> http2InputMessage =
new Http2InputMessageFrame<>(new FinalFragmentByteArrayInputStream(messagePart.array(), last), false);
webSocketTransportListener.onData(http2InputMessage);
}
}
25 changes: 18 additions & 7 deletions ...e-websocket/src/main/java/org/apache/dubbo/rpc/protocol/tri/websocket/TripleEndpoint.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.dubbo.remoting.http12.HttpHeaders;
import org.apache.dubbo.remoting.http12.HttpMethods;
import org.apache.dubbo.remoting.http12.HttpStatus;
import org.apache.dubbo.remoting.http12.MessageTypeToken;
import org.apache.dubbo.remoting.http12.h2.Http2Header;
import org.apache.dubbo.remoting.http12.h2.Http2InputMessage;
import org.apache.dubbo.remoting.http12.h2.Http2InputMessageFrame;
Expand All @@ -39,6 +40,9 @@
import javax.websocket.EndpointConfig;
import javax.websocket.Session;

import java.io.InputStream;
import java.io.OutputStream;

import static org.apache.dubbo.rpc.protocol.tri.websocket.WebSocketConstants.TRIPLE_WEBSOCKET_LISTENER;

public class TripleEndpoint extends Endpoint {
Expand All @@ -55,36 +59,43 @@ public void onOpen(Session session, EndpointConfig config) {
TripleConfig tripleConfig = ConfigManager.getProtocolOrDefault(url).getTripleOrDefault();

WebSocketStreamChannel webSocketStreamChannel = new WebSocketStreamChannel(session, tripleConfig);
WebSocketTransportListener webSocketTransportListener =
WebSocketTransportListener<InputStream, OutputStream> webSocketTransportListener =
DefaultWebSocketServerTransportListenerFactory.INSTANCE.newInstance(
webSocketStreamChannel, url, FrameworkModel.defaultModel());
webSocketStreamChannel,
url,
FrameworkModel.defaultModel(),
new MessageTypeToken<InputStream, OutputStream>() {});
webSocketTransportListener.onMetadata(http2Header);
session.addMessageHandler(new TripleTextMessageHandler(webSocketTransportListener));
session.addMessageHandler(new TripleBinaryMessageHandler(webSocketTransportListener));
session.getUserProperties().put(TRIPLE_WEBSOCKET_LISTENER, webSocketTransportListener);
}

@SuppressWarnings("unchecked")
@Override
public void onClose(Session session, CloseReason closeReason) {
super.onClose(session, closeReason);
WebSocketTransportListener webSocketTransportListener =
(WebSocketTransportListener) session.getUserProperties().get(TRIPLE_WEBSOCKET_LISTENER);
WebSocketTransportListener<InputStream, OutputStream> webSocketTransportListener =
(WebSocketTransportListener<InputStream, OutputStream>)
session.getUserProperties().get(TRIPLE_WEBSOCKET_LISTENER);
if (webSocketTransportListener == null) {
return;
}
if (closeReason.getCloseCode().getCode() == CloseCodes.NORMAL_CLOSURE.getCode()) {
Http2InputMessage http2InputMessage = new Http2InputMessageFrame(StreamUtils.EMPTY, true);
Http2InputMessage<InputStream> http2InputMessage = new Http2InputMessageFrame<>(StreamUtils.EMPTY, true);
webSocketTransportListener.onData(http2InputMessage);
return;
}
webSocketTransportListener.cancelByRemote(closeReason.getCloseCode().getCode());
}

@SuppressWarnings("unchecked")
@Override
public void onError(Session session, Throwable thr) {
super.onError(session, thr);
WebSocketTransportListener webSocketTransportListener =
(WebSocketTransportListener) session.getUserProperties().get(TRIPLE_WEBSOCKET_LISTENER);
WebSocketTransportListener<InputStream, OutputStream> webSocketTransportListener =
(WebSocketTransportListener<InputStream, OutputStream>)
session.getUserProperties().get(TRIPLE_WEBSOCKET_LISTENER);
if (webSocketTransportListener == null) {
return;
}
Expand Down
8 changes: 5 additions & 3 deletions ...t/src/main/java/org/apache/dubbo/rpc/protocol/tri/websocket/TripleTextMessageHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,19 +23,21 @@

import javax.websocket.MessageHandler;

import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;

public class TripleTextMessageHandler implements MessageHandler.Partial<String> {

private final WebSocketTransportListener webSocketTransportListener;
private final WebSocketTransportListener<InputStream, OutputStream> webSocketTransportListener;

public TripleTextMessageHandler(WebSocketTransportListener webSocketTransportListener) {
public TripleTextMessageHandler(WebSocketTransportListener<InputStream, OutputStream> webSocketTransportListener) {
this.webSocketTransportListener = webSocketTransportListener;
}

@Override
public void onMessage(String messagePart, boolean last) {
Http2InputMessage http2InputMessage = new Http2InputMessageFrame(
Http2InputMessage<InputStream> http2InputMessage = new Http2InputMessageFrame<>(
new FinalFragmentByteArrayInputStream(messagePart.getBytes(StandardCharsets.UTF_8), last), false);
webSocketTransportListener.onData(http2InputMessage);
}
Expand Down
9 changes: 5 additions & 4 deletions ...ket/src/main/java/org/apache/dubbo/rpc/protocol/tri/websocket/WebSocketStreamChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
Expand All @@ -44,7 +45,7 @@

import static org.apache.dubbo.rpc.protocol.tri.websocket.WebSocketConstants.TRIPLE_WEBSOCKET_REMOTE_ADDRESS;

public class WebSocketStreamChannel implements H2StreamChannel {
public class WebSocketStreamChannel implements H2StreamChannel<OutputStream> {

private final Session session;

Expand Down Expand Up @@ -78,8 +79,8 @@ public CompletableFuture<Void> writeResetFrame(long errorCode) {
}

@Override
public Http2OutputMessage newOutputMessage(boolean endStream) {
return new Http2OutputMessageFrame(
public Http2OutputMessage<OutputStream> newOutputMessage(boolean endStream) {
return new Http2OutputMessageFrame<>(
new LimitedByteArrayOutputStream(256, tripleConfig.getMaxResponseBodySizeOrDefault()), endStream);
}

Expand All @@ -99,7 +100,7 @@ public CompletableFuture<Void> writeHeader(HttpMetadata httpMetadata) {
}

@Override
public CompletableFuture<Void> writeMessage(HttpOutputMessage httpOutputMessage) {
public CompletableFuture<Void> writeMessage(HttpOutputMessage<OutputStream> httpOutputMessage) {
ByteArrayOutputStream body = (ByteArrayOutputStream) httpOutputMessage.getBody();
CompletableFuture<Void> completableFuture = new CompletableFuture<>();
try {
Expand Down
6 changes: 6 additions & 0 deletions dubbo-remoting/dubbo-remoting-http12/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,12 @@
<artifactId>netty-codec-http2</artifactId>
</dependency>

<dependency>
<groupId>org.apache.fory</groupId>
<artifactId>fory-core</artifactId>
<version>0.12.0</version>
</dependency>

<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java-util</artifactId>
Expand Down
21 changes: 9 additions & 12 deletions ...pc/protocol/tri/frame/RecordListener.java → ...oting/http12/EmptyInputStreamMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,21 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc.protocol.tri.frame;
package org.apache.dubbo.remoting.http12;

public class RecordListener implements TriDecoder.Listener {
byte[] lastData;
int dataCount;
boolean close;
import org.apache.dubbo.common.io.StreamUtils;

@Override
public void onRawMessage(byte[] data) {
dataCount += 1;
lastData = data;
}
import java.io.InputStream;

public class EmptyInputStreamMessage implements HttpInputMessage<InputStream> {

public static final EmptyInputStreamMessage INSTANCE = new EmptyInputStreamMessage();

@Override
public void close() {
close = true;
public InputStream getBody() {
return StreamUtils.EMPTY;
}
}
Loading
Loading