BitoAgent commented on code in PR #13786:
URL: https://github.com/apache/dubbo/pull/13786#discussion_r1568049600
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodec.java:
##########
@@ -16,123 +16,103 @@
*/
package org.apache.dubbo.rpc.protocol.tri.h12.grpc;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.io.StreamUtils;
+import org.apache.dubbo.common.utils.ArrayUtils;
import org.apache.dubbo.remoting.http12.exception.DecodeException;
import org.apache.dubbo.remoting.http12.exception.EncodeException;
import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
import org.apache.dubbo.remoting.http12.message.MediaType;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+import org.apache.dubbo.rpc.model.MethodDescriptor;
+import org.apache.dubbo.rpc.model.PackableMethod;
+import org.apache.dubbo.rpc.model.PackableMethodFactory;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
-import com.google.protobuf.Message;
-
-import static
org.apache.dubbo.common.constants.CommonConstants.PROTOBUF_MESSAGE_CLASS_NAME;
+import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
+import static
org.apache.dubbo.common.constants.CommonConstants.DUBBO_PACKABLE_METHOD_FACTORY;
public class GrpcCompositeCodec implements HttpMessageCodec {
- private final ProtobufHttpMessageCodec protobufHttpMessageCodec;
+ private static final String PACKABLE_METHOD_CACHE =
"PACKABLE_METHOD_CACHE";
- private final WrapperHttpMessageCodec wrapperHttpMessageCodec;
+ private final URL url;
- public GrpcCompositeCodec(
- ProtobufHttpMessageCodec protobufHttpMessageCodec,
WrapperHttpMessageCodec wrapperHttpMessageCodec) {
- this.protobufHttpMessageCodec = protobufHttpMessageCodec;
- this.wrapperHttpMessageCodec = wrapperHttpMessageCodec;
- }
+ private final FrameworkModel frameworkModel;
+
+ private final String mediaType;
- public void setEncodeTypes(Class<?>[] encodeTypes) {
- this.wrapperHttpMessageCodec.setEncodeTypes(encodeTypes);
+ private PackableMethod packableMethod;
+
+ public GrpcCompositeCodec(URL url, FrameworkModel frameworkModel, String
mediaType) {
+ this.url = url;
+ this.frameworkModel = frameworkModel;
+ this.mediaType = mediaType;
}
- public void setDecodeTypes(Class<?>[] decodeTypes) {
- this.wrapperHttpMessageCodec.setDecodeTypes(decodeTypes);
+ public void loadPackableMethod(MethodDescriptor methodDescriptor) {
+ if (methodDescriptor instanceof PackableMethod) {
+ packableMethod = (PackableMethod) methodDescriptor;
+ return;
+ }
+ Map<MethodDescriptor, PackableMethod> cacheMap =
(Map<MethodDescriptor, PackableMethod>) url.getServiceModel()
+ .getServiceMetadata()
+ .getAttributeMap()
+ .computeIfAbsent(PACKABLE_METHOD_CACHE, k -> new
ConcurrentHashMap<>());
+ packableMethod = cacheMap.computeIfAbsent(methodDescriptor, md ->
frameworkModel
+ .getExtensionLoader(PackableMethodFactory.class)
+
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel())
+ .getString(DUBBO_PACKABLE_METHOD_FACTORY, DEFAULT_KEY))
+ .create(methodDescriptor, url, mediaType));
}
@Override
public void encode(OutputStream outputStream, Object data, Charset
charset) throws EncodeException {
- // protobuf
- // TODO int compressed =
Identity.MESSAGE_ENCODING.equals(requestMetadata.compressor.getMessageEncoding())
? 0 :
- // 1;
try {
- int compressed = 0;
- outputStream.write(compressed);
- if (isProtobuf(data)) {
- ProtobufWriter.write(protobufHttpMessageCodec, outputStream,
data);
- return;
- }
- // wrapper
- wrapperHttpMessageCodec.encode(outputStream, data);
- } catch (IOException e) {
+ outputStream.write(0);
+ byte[] bytes = packableMethod.packResponse(data);
+ writeLength(outputStream, bytes.length);
+ outputStream.write(bytes);
+ } catch (Exception e) {
throw new EncodeException(e);
}
}
@Override
public Object decode(InputStream inputStream, Class<?> targetType, Charset
charset) throws DecodeException {
- if (isProtoClass(targetType)) {
- return protobufHttpMessageCodec.decode(inputStream, targetType,
charset);
+ try {
+ byte[] data = StreamUtils.readBytes(inputStream);
Review Comment:
**Optimization Issue**: Reading the entire InputStream into a byte array
for processing may lead to increased memory usage, especially for large data
sizes. This can affect the performance and scalability of the application. <br>
**Fix**: Consider processing the InputStream directly without converting it to
a byte array. If the parsing library supports it, use methods that read from an
InputStream directly. This approach reduces memory usage and can improve
performance. <br> **Code Suggestion**:
```
+ return packableMethod.parseRequestDirectly(inputStream);
```
##########
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/h12/grpc/GrpcCompositeCodec.java:
##########
@@ -16,123 +16,103 @@
*/
package org.apache.dubbo.rpc.protocol.tri.h12.grpc;
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.config.ConfigurationUtils;
+import org.apache.dubbo.common.io.StreamUtils;
+import org.apache.dubbo.common.utils.ArrayUtils;
import org.apache.dubbo.remoting.http12.exception.DecodeException;
import org.apache.dubbo.remoting.http12.exception.EncodeException;
import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
import org.apache.dubbo.remoting.http12.message.MediaType;
+import org.apache.dubbo.rpc.model.FrameworkModel;
+import org.apache.dubbo.rpc.model.MethodDescriptor;
+import org.apache.dubbo.rpc.model.PackableMethod;
+import org.apache.dubbo.rpc.model.PackableMethodFactory;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.Charset;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
-import com.google.protobuf.Message;
-
-import static
org.apache.dubbo.common.constants.CommonConstants.PROTOBUF_MESSAGE_CLASS_NAME;
+import static org.apache.dubbo.common.constants.CommonConstants.DEFAULT_KEY;
+import static
org.apache.dubbo.common.constants.CommonConstants.DUBBO_PACKABLE_METHOD_FACTORY;
public class GrpcCompositeCodec implements HttpMessageCodec {
- private final ProtobufHttpMessageCodec protobufHttpMessageCodec;
+ private static final String PACKABLE_METHOD_CACHE =
"PACKABLE_METHOD_CACHE";
- private final WrapperHttpMessageCodec wrapperHttpMessageCodec;
+ private final URL url;
- public GrpcCompositeCodec(
- ProtobufHttpMessageCodec protobufHttpMessageCodec,
WrapperHttpMessageCodec wrapperHttpMessageCodec) {
- this.protobufHttpMessageCodec = protobufHttpMessageCodec;
- this.wrapperHttpMessageCodec = wrapperHttpMessageCodec;
- }
+ private final FrameworkModel frameworkModel;
+
+ private final String mediaType;
- public void setEncodeTypes(Class<?>[] encodeTypes) {
- this.wrapperHttpMessageCodec.setEncodeTypes(encodeTypes);
+ private PackableMethod packableMethod;
+
+ public GrpcCompositeCodec(URL url, FrameworkModel frameworkModel, String
mediaType) {
+ this.url = url;
+ this.frameworkModel = frameworkModel;
+ this.mediaType = mediaType;
}
- public void setDecodeTypes(Class<?>[] decodeTypes) {
- this.wrapperHttpMessageCodec.setDecodeTypes(decodeTypes);
+ public void loadPackableMethod(MethodDescriptor methodDescriptor) {
+ if (methodDescriptor instanceof PackableMethod) {
+ packableMethod = (PackableMethod) methodDescriptor;
+ return;
+ }
+ Map<MethodDescriptor, PackableMethod> cacheMap =
(Map<MethodDescriptor, PackableMethod>) url.getServiceModel()
+ .getServiceMetadata()
+ .getAttributeMap()
+ .computeIfAbsent(PACKABLE_METHOD_CACHE, k -> new
ConcurrentHashMap<>());
+ packableMethod = cacheMap.computeIfAbsent(methodDescriptor, md ->
frameworkModel
+ .getExtensionLoader(PackableMethodFactory.class)
+
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel())
+ .getString(DUBBO_PACKABLE_METHOD_FACTORY, DEFAULT_KEY))
+ .create(methodDescriptor, url, mediaType));
Review Comment:
**Optimization Issue**: The use of ConcurrentHashMap.computeIfAbsent within
loadPackableMethod could lead to unnecessary synchronization overhead and
potential contention under high-load scenarios, affecting performance. <br>
**Fix**: Consider using a simple check-then-act idiom with
ConcurrentHashMap.get() followed by ConcurrentHashMap.putIfAbsent() to reduce
synchronization overhead. This approach minimizes locking duration, especially
if the PackableMethod is already computed and present in the cache. <br> **Code
Suggestion**:
```
+ PackableMethod result = cacheMap.get(methodDescriptor);
+ if (result == null) {
+ PackableMethod newValue = frameworkModel
+ .getExtensionLoader(PackableMethodFactory.class)
+
.getExtension(ConfigurationUtils.getGlobalConfiguration(url.getApplicationModel())
+ .getString(DUBBO_PACKABLE_METHOD_FACTORY,
DEFAULT_KEY))
+ .create(methodDescriptor, url, mediaType);
+ result = cacheMap.putIfAbsent(methodDescriptor, newValue);
+ if (result == null) {
+ result = newValue;
+ }
+ }
+ packableMethod = result;
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
