[REEF-362] Implement a Wake Transport using HTTP #1341
Conversation
Additional implementation for Wake via HTTP JIRA: [REEF-362](https://issues.apache.org/jira/browse/REEF-362) Pull Request This closes #
|
@nhne How about adding tests on |
|
Added one Test on |
|
@nhne I've skimmed through this PR, but it looks like it has a bunch of duplicated codes. Can't we minimize the duplicated codes? Maybe we need to refactor the original code to do this. |
|
I implmeneted in this way to avoid modify the original code. I just worried I can insert err code into original code and compatibility issues. I think refactoring the original code can be eliminate many duplicated code since It does not have much difference. What Do you think would be better? |
Refactored Netty Messaging Transportand Netty Link and Listeners. but still have to change implmentation of MessagingTransportFactory
| /** | ||
| * Option for use http. | ||
| */ | ||
| @NamedParameter(doc = "Option for use http.", default_value = "" + PROTOCOL_NETTY) |
There was a problem hiding this comment.
Converting String using blank string and + is a bad practice. Use String.valueOf instead.
There was a problem hiding this comment.
It was used across the original source. Is it better to update to String.valueOf every use?
There was a problem hiding this comment.
It would be better to make another issue for refactoring the whole project.
| * @return transport | ||
| */ | ||
| Transport newInstance(final String hostAddress, | ||
| int port, |
There was a problem hiding this comment.
It would be refactored during another PR. Thank you!
There was a problem hiding this comment.
You don't have to refactor it in another PR because it is a new code. Please just add final here.
| final int retryTimeout, | ||
| final int protocol) { | ||
| try { | ||
| TcpPortProvider tcpPortProvider = Tang.Factory.getTang().newInjector().getInstance(TcpPortProvider.class); |
| import io.netty.channel.ChannelFutureListener; | ||
| import org.apache.reef.wake.remote.transport.LinkListener; | ||
|
|
||
| class NettyChannelFutureListener<T> implements ChannelFutureListener { |
There was a problem hiding this comment.
This class should be public & final if there is no class extending it.
| /** | ||
| * the buffer size of the frame decoder. | ||
| */ | ||
| public static final int MAXFRAMELENGTH = 10 * 1024 * 1024; | ||
| private final NettyChannelHandlerFactory handlerFactory; | ||
| private final SslContext sslContext; |
There was a problem hiding this comment.
I think explanations on these variables are necessary.
| } | ||
|
|
||
| final Channel channel = ctx.channel(); | ||
| byte[] message = new byte[content.readableBytes()]; |
| } | ||
|
|
||
| @Override | ||
| protected void exceptionCleanup(final ChannelHandlerContext ctx, final Throwable cause) { |
There was a problem hiding this comment.
It is required method to extend AbstractNettyEventyListener and it only performs close channel when it is clientListener.
|
|
||
| if (this.uri != null) { | ||
| try { | ||
| FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, uri.getRawPath()); |
| if (listener != null) { | ||
| future.addListener(new NettyChannelFutureListener<>(message, listener)); | ||
| } | ||
| } catch (InterruptedException ex) { |
| private final AbstractNettyEventListener clientEventListener; | ||
| private final AbstractNettyEventListener serverEventListener; | ||
|
|
||
| private final boolean ssl = System.getProperty("ssl") != null; |
There was a problem hiding this comment.
I don't think initializing final variable here is good, because it is not a static variable. How about moving this into the constructor?
|
@DifferentSC Thank you for reviewing my work! I'll check the things you reviewed. |
| * Unique protocol numbers for choosing protocols. | ||
| */ | ||
| public static final int PROTOCOL_NETTY = 100; | ||
| public static final int PROTOCOL_HTTP = 101; |
There was a problem hiding this comment.
We should probably use enum here.
There was a problem hiding this comment.
Unfortunately, @DifferentSC said that using enum which is related to Tang Configuration might be not supported for now. Instead, I suggest it could be changed to static final String.
There was a problem hiding this comment.
@nhne You can convert enum to string. Please see https://stackoverflow.com/questions/6667243/using-enum-values-as-string-literals
| Transport newInstance(final int port, | ||
| final EventHandler<TransportEvent> clientHandler, | ||
| final EventHandler<TransportEvent> serverHandler, | ||
| final EventHandler<Exception> exHandler); |
There was a problem hiding this comment.
I love my variables final, but we probably should not mix functional and non-functional updates in one PR. Next time please submit your stylistic fixes in a separate pull request (and yes, we welcome such updates!) :)
| @@ -93,7 +93,7 @@ public Transport newInstance(final String hostAddress, | |||
| final int numberOfTries, | |||
| final int retryTimeout) { | |||
| try { | |||
| TcpPortProvider tcpPortProvider = Tang.Factory.getTang().newInjector().getInstance(TcpPortProvider.class); | |||
| final TcpPortProvider tcpPortProvider = Tang.Factory.getTang().newInjector().getInstance(TcpPortProvider.class); | |||
There was a problem hiding this comment.
Tang.Factory.getTang() called many times in this class. Can declare
private static final Tang TANG = Tang.Factory.getTang();|
|
||
| private static final int HANDLER_NETTY = 100; | ||
| private static final int HANDLER_HTTP_SERVER = 101; | ||
| private static final int HANDLER_HTTP_CLIENT = 102; |
| final class NettyHttpClientEventListener extends AbstractNettyEventListener { | ||
|
|
||
| private static final Logger LOG = Logger.getLogger(NettyHttpClientEventListener.class.getName()); | ||
| private StringBuilder buf = new StringBuilder(); |
| new InetSocketAddress(hostAddress, port), | ||
| new ObjectSerializableCodec<String>(), | ||
| new LoggingLinkListener<String>()); | ||
| link.write(new String("hello1")); |
There was a problem hiding this comment.
no need for new String(...) - "hello1" is a string already
There was a problem hiding this comment.
The test was pasted from former TransportTest. This whole test would be changed after refactoring code. Thank you!
| new ObjectSerializableCodec<String>(), | ||
| new LoggingLinkListener<String>()); | ||
| link.write(new String("hello1")); | ||
| link.write(new String("hello2")); |
|
|
||
| @Test | ||
| public void testHttpTransportTestEvent() throws Exception { | ||
| System.out.println(LOG_PREFIX + name.getMethodName()); |
| link.write(new TestEvent("hello2", 1.0)); | ||
|
|
||
| monitor.mwait(); | ||
| transport.close(); |
There was a problem hiding this comment.
Transport implements AutoCloseable - can use it in try block instead of calling .close() explicitly
| private final Codec<T> codec; | ||
| private final Monitor monitor; | ||
| private final int expected; | ||
| private AtomicInteger count = new AtomicInteger(0); |
|
@motus I appreciate for your kind and precise review! I'll check out all about these. thanks a lot! |
added a java doc and fixed format error
| @@ -108,23 +150,3 @@ public String toString() { | |||
| return "NettyLink: " + channel; // Channel has good .toString() implementation | |||
| } | |||
| } | |||
|
|
|||
There was a problem hiding this comment.
It is separted into another file "NettyChannelFutureListener.java"
| final EStage<TransportEvent> serverStage, | ||
| final int numberOfTries, | ||
| final int retryTimeout, | ||
| final String protocol); |
| final EStage<TransportEvent> serverStage, | ||
| final int numberOfTries, | ||
| final int retryTimeout, | ||
| final String protocol) { |
There was a problem hiding this comment.
this variable is not used
| * @return transport | ||
| */ | ||
| Transport newInstance(final String hostAddress, | ||
| int port, |
There was a problem hiding this comment.
You don't have to refactor it in another PR because it is a new code. Please just add final here.
| public final class NettyChannelFutureListener<T> implements ChannelFutureListener { | ||
|
|
||
| private final T message; | ||
| private LinkListener<T> listener; |
|
|
||
| /* Types for initiating channel */ | ||
| public enum ChannelType { | ||
| NETTY, HTTP_SERVER, HTTP_CLIENT |
There was a problem hiding this comment.
What is NETTY? Is it a tcp or udp socket channel? The naming is unclear to me because HTTP server and client also uses Netty.
| import java.util.logging.Logger; | ||
|
|
||
| /** | ||
| * A Netty event listener for server. |
There was a problem hiding this comment.
Oops, my mistake! Thank you!
| this.httpRequest = request; | ||
|
|
||
| if (!headers.isEmpty()) { | ||
| for (final Map.Entry<String, String> h: headers) { |
| } | ||
| buf.append("\r\n"); | ||
| } | ||
|
|
There was a problem hiding this comment.
Same as above. Don't we have to create a new StringBuillder here?
| if (listener != null) { | ||
| future.addListener(new NettyChannelFutureListener<>(message, listener)); | ||
|
|
||
| if (this.uri != null) { |
There was a problem hiding this comment.
I think creating a NettyHttpLink would be better instead of fixing this class.
We have to check whenever .write is called whether the uri is null or not. This doesn't look good to me.
| this.clientEventListener = new NettyClientEventListener(this.addrToLinkRefMap, clientStage); | ||
| this.serverEventListener = new NettyServerEventListener(this.addrToLinkRefMap, serverStage); | ||
| if (protocolType.equals(PROTOCOL_NETTY)) { | ||
| this.clientEventListener = new NettyClientEventListener(this.addrToLinkRefMap, clientStage); |
There was a problem hiding this comment.
How about receiving client and server event listener from the constructor?
We can remove if ... else if we inject them in the constructor.
There was a problem hiding this comment.
I've discussed it with @nhne in offline, and we decided not to refactor the constructor because there is a little benefit of refactoring it.
|
@taegeonum Thanks for sharing detailed review! I'll check all you mentioned above. Thank you. |
| /** | ||
| * Factory that creates a NettyLink. | ||
| */ | ||
| public final class NettyDefaultLinkFactory<T> implements NettyLinkFactory { |
There was a problem hiding this comment.
Why don't you create HttpLinkFactory?
There was a problem hiding this comment.
I've created NettyHttpLinkFactory. Thank you!
| * @param listener the listener | ||
| */ | ||
| Link<T> newInstance(final Channel channel, | ||
| final Encoder<? super T> encoder, |
|
|
||
| @Override | ||
| public String toString() { | ||
| return "NettyLink: " + channel; // Channel has good .toString() implementation |
There was a problem hiding this comment.
Please use a StringBuffer
| * If you set a {@code LinkListener<T>}, it keeps message until writeAndFlush operation completes | ||
| * and notifies whether the sent message transferred successfully through the listener. | ||
| */ | ||
| public class NettyHttpLink<T> implements Link<T> { |
| buf.setLength(0); // clearing the buffer | ||
| } | ||
|
|
||
| if (message.length > 0) { |
There was a problem hiding this comment.
Could you explain when does the message length is zero? and don't we send the message if the length is zero?
There was a problem hiding this comment.
By currrent implementation, Server does not send any response to client. Actually NettyHttpEventListener does not receive any response now. But if server return the response to client, there can be HttpResponse with empty message.
We can choose to clear the channelRead method to do nothing or leave it for latter usage.
| */ | ||
| public static final String PROTOCOL_TCP = ProtocolTypes.TCP.name(); | ||
| public static final String PROTOCOL_HTTP = ProtocolTypes.HTTP.name(); | ||
| public static final String PROTOCOL_HTTPS = ProtocolTypes.HTTPS.name(); |
There was a problem hiding this comment.
Do we handle supporting https in this PR?
There was a problem hiding this comment.
I Think features about HTTPS should be handled in another PR. Thank you!
|
Thanks for your review! @taegeonum. I'll reconsider the structure of Http Channels considering your review. |
|
@taegeonum Sorry for very late commit! I have addressed the review. Would you please have a look? |
| if(isServer) { | ||
| pipeline | ||
| .addLast("codec", new HttpServerCodec()) | ||
| .addLast("requestDecoder", new HttpRequestDecoder()) |
There was a problem hiding this comment.
Why do we need both HttpServerCodec and HttpRequestDecoder/Encoder?
There was a problem hiding this comment.
I followed example in the Netty Http server. But by reference document, I think we do need additional codec since HttpServerCodec is combination of them. I'll remove and test about it.
| default: | ||
| throw new IllegalArgumentException("Invalid type of channel"); | ||
| } | ||
| // every channel's pipeline have a same inbound handler. |
There was a problem hiding this comment.
every channel pipeline has the same inbound handler
| this.stage.onNext(this.getTransportEvent(message, channel)); | ||
| } | ||
| } else { | ||
| LOG.log(Level.SEVERE, "Unknown type of message received: {0}", msg); |
|
|
||
| content.readBytes(message); | ||
| if (LOG.isLoggable(Level.FINEST)) { | ||
| buf.append("CONTENT: ").append(content.toString(CharsetUtil.UTF_8)).append("\r\n"); |
There was a problem hiding this comment.
Why don't you define buf as a local variable? It doesn't look like this should be a member variable.
| .set(HttpHeaders.Names.CONTENT_LENGTH, buf.readableBytes()); | ||
| request.content().clear().writeBytes(buf); | ||
| final ChannelFuture future = channel.writeAndFlush(request); | ||
| future.sync(); |
There was a problem hiding this comment.
I wanted to throw exception when the result of future is failure by sync(). but i found its order is weird because listener must be added before its sync!
| */ | ||
| final class NettyHttpServerEventListener extends AbstractNettyEventListener { | ||
|
|
||
| private final StringBuilder buf = new StringBuilder(); |
There was a problem hiding this comment.
Same as above. Do we have to define it as a member variable?
|
My former implementation need buf as global variable, but now we don't have to keep it global, so I converted it into local. |
| final ByteBuf content = httpContent.content(); | ||
| final Channel channel = ctx.channel(); | ||
| final byte[] message = new byte[content.readableBytes()]; | ||
| final StringBuilder buf = new StringBuilder(); |
There was a problem hiding this comment.
please put this variable inside the following if block (line 62)
| } | ||
|
|
||
| @Override | ||
| public void channelRead(final ChannelHandlerContext ctx, final Object msg) { |
There was a problem hiding this comment.
How about the following code?
public void channelRead(final ChannelHandlerContext ctx, final FullHttpResponse response) {
byte[] content = null;
ByteBuf byteBuf = response.content();
if (byteBuf.hasArray()) {
content = byteBuf.array();
} else {
content = new byte[byteBuf.readableBytes()];
byteBuf.readBytes(content);
}
....
}There was a problem hiding this comment.
I'm sorry that as method channelRead is overriding method, I cannot change type of parameter.
But i'll adapt the name change and other things. Thank you!
| } | ||
|
|
||
| @Override | ||
| public void channelRead(final ChannelHandlerContext ctx, final Object msg) { |
There was a problem hiding this comment.
Same as the ClientEventListener codes
logging buffer's behavior has changed, content variable also changed
|
Sorry for late check. I have committed about your review. Thank you very much! @taegeonum |
| channel.localAddress(), channel.remoteAddress(), buf}); | ||
| } | ||
|
|
||
| if (content.length > 0) { |
| channel.localAddress(), channel.remoteAddress(), byteBuf}); | ||
| } | ||
|
|
||
| if (content.length > 0) { |
|
I thought about this somewhile... If any listener expect any response without any parameter, intetionally empty request can be sent. Then, this check would be erroneous. So I removed the if statements. Thank you! @taegeonum |
| if (listener != null) { | ||
| future.addListener(new NettyChannelFutureListener<>(message, listener)); | ||
| } | ||
| future.sync(); |
There was a problem hiding this comment.
It is because sync() will throw an Exception when sending request is Interrupted. To catch a failure on sending request, sync was used here.
There was a problem hiding this comment.
I mean, why do we need sync? you already registered the future listener and it doesn't look we need sync()
There was a problem hiding this comment.
listener can be null in constructor. If there is no listener attached, there would be no warning about failure on sending request. In my opinion, if default listener for failure detection is provided, we don't have to use sync().
There was a problem hiding this comment.
Using Future and listener is to asynchronously write the requests, but if you do sync() immediately, I think there is no benefit of the asynchronous write.
There was a problem hiding this comment.
I strongly agree with you. Then what about change the implmentation to adopt default listener for detect failure on complete? or just delete sync()? Can I ask which would be better choice?
There was a problem hiding this comment.
Sorry, I don't fully understand your question. You've already registered the future listener in line 104.
There was a problem hiding this comment.
@taegeonum the listener can be null. The constructor of NettyHttpLink at line59 makes listener null. So There can be no listener for catching failure if sync() is not performed.
There was a problem hiding this comment.
@nhne I see. I think it is okay to me. If we want to ignore the message futures, we can skip the listener. Otherwise, we provide the listener and handle the message futures. So, how about just removing sync()?
There was a problem hiding this comment.
Then I'll remove sync(). User may provide listener if they need any failure detection. Thank you for sharing your opinion!
| try { | ||
| final FullHttpRequest request = | ||
| new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, uri.getRawPath()); | ||
| final ByteBuf buf = Unpooled.copiedBuffer(encoder.encode(message)); |
There was a problem hiding this comment.
Why don't you use wrappedBuffer? copiedBuffer performs additional byte copy.
|
@taegeonum I have changed copiedBuffer into wrappedBuffer. Also I commented about sync upper. |
|
@taegeonum Thank you for sharing your idea for |
| */ | ||
| @Override | ||
| public void write(final T message) { | ||
| LOG.log(Level.FINEST, "write {0} :: {1}", new Object[] {channel, message}); |
There was a problem hiding this comment.
Could you please add the following code?
if (LOG.isLoggable(Level.FINEST)) {
|
@taegeonum fixed the |
|
|
||
| // for HTTP and default Netty | ||
| if (protocolType == ProtocolType.HTTP) { | ||
| this.uri = URI.create("http://" + hostAddress); |
There was a problem hiding this comment.
@nhne hostAddress -> host
I've got an exception message when I tested it on my local machine, even though the test passed
Caused by: java.lang.IllegalArgumentException: Illegal character in fragment at index 8: http://##UNKNOWN##
at java.net.URI.create(URI.java:852)
at org.apache.reef.wake.remote.transport.netty.NettyMessagingTransport.<init>(NettyMessagingTransport.java:139)
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
at org.apache.reef.tang.implementation.java.InjectorImpl.injectFromPlan(InjectorImpl.java:637)
... 41 more
There was a problem hiding this comment.
Sorry for late check. I had to reconfigure my environment to reproduce the error. I'll fix it and push a commit now!
|
@taegeonum I have fixed |
| @@ -139,7 +139,7 @@ private NettyMessagingTransport( | |||
| try{ | |||
| this.uri = URI.create("http://" + host); | |||
| } catch (IllegalArgumentException e){ | |||
There was a problem hiding this comment.
Could you please add final ?
|
@nhne LGTM except for the last comment :) |
|
@taegeonum I have pushed a commit reflecting your review. Thank you! |
|
@nhne @taegeonum Thanks for working on this issue. Could you also make sure that this PR works correctly in distributed environments? |
Additional implementation for Wake via HTTP
new packageorg.apache.reef.wake.remote.transport.netty.httpJIRA:
REEF-362
Pull Request
This closes #1341