Skip to content

[WIP]Upgrade Netty to 4.1.17 #19829

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

Closed
zsxwing wants to merge 3 commits into from
Closed

[WIP]Upgrade Netty to 4.1.17 #19829

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
14dbc71
Upgrade Netty to 4.1.17
zsxwing Nov 27, 2017
93632a0
Fix ProtocolSuite
zsxwing Nov 27, 2017
96df5f2
Add AbstractFileRegion
zsxwing Dec 9, 2017
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
57 changes: 57 additions & 0 deletions common/network-common/src/main/java/org/apache/spark/network/crypto/TransportCipher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,6 +203,63 @@ public long transfered() {
return transferred;
}

@Override

@vanzin vanzin Dec 6, 2017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I prototyped this change internally for a different reason, and I chose to create an abstract class for all this boilerplate, which seems cleaner to me.

public abstract class AbstractFileRegion extends AbstractReferenceCounted implements FileRegion {

  @Override
  @SuppressWarnings("deprecation")
  public final long transfered() {
    return transferred();
  }

  @Override
  public FileRegion retain() {
    super.retain();
    return this;
  }

  @Override
  public FileRegion retain(int increment) {
    super.retain(increment);
    return this;
  }

  @Override
  public FileRegion touch() {
    return touch(null);
  }

  @Override
  public FileRegion touch(Object hint) {
    return this;
  }

}

@zsxwing zsxwing Dec 9, 2017

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point! Updated!

public long transferred() {
return transferred;
}

/**
* Override this due to different return types of ReferenceCounted.touch and FileRegion.touch.
*/
@Override
public EncryptedMessage touch() {
super.touch();
return this;
}

@Override
public EncryptedMessage touch(Object o) {
if (region != null) {
region.touch(o);
}
if (buf != null) {
buf.touch(o);
}
return this;
}

/**
* Override this due to different return types of ReferenceCounted.touch and FileRegion.touch.
*/
@Override
public EncryptedMessage retain() {
super.retain();
return this;
}

@Override
public EncryptedMessage retain(int increment) {
super.retain(increment);
if (region != null) {
region.retain(increment);
}
if (buf != null) {
buf.retain(increment);
}
return this;
}

@Override
public boolean release(int decrement) {
if (region != null) {
region.release(decrement);
}
if (buf != null) {
buf.release(decrement);
}
return super.release(decrement);
}

@Override
public long transferTo(WritableByteChannel target, long position) throws IOException {
Preconditions.checkArgument(position == transfered(), "Invalid position.");
Expand Down
51 changes: 51 additions & 0 deletions common/network-common/src/main/java/org/apache/spark/network/protocol/MessageWithHeader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,11 @@ public long transfered() {
return totalBytesTransferred;
}

@Override
public long transferred() {
return totalBytesTransferred;
}

/**
* This code is more complicated than you would think because we might require multiple
* transferTo invocations in order to transfer a single MessageWithHeader to avoid busy waiting.
Expand Down Expand Up @@ -160,4 +165,50 @@ private int writeNioBuffer(

return ret;
}

/** Override this due to different return types of ReferenceCounted.touch and FileRegion.touch. */
@Override
public MessageWithHeader touch() {
super.touch();
return this;
}

@Override
public MessageWithHeader touch(Object o) {
header.touch(o);
ReferenceCountUtil.touch(body, o);
return this;
}

/** Override this due to different return types of ReferenceCounted.touch and FileRegion.touch. */
@Override
public MessageWithHeader retain() {
super.retain();
return this;
}

@Override
public MessageWithHeader retain(int increment) {
super.retain(increment);
header.retain(increment);
ReferenceCountUtil.retain(body, increment);
if (managedBuffer != null) {
for (int i = 0; i < increment; i++) {
managedBuffer.retain();
}
}
return this;
}

@Override
public boolean release(int decrement) {
header.release(decrement);
ReferenceCountUtil.release(body, decrement);
if (managedBuffer != null) {
for (int i = 0; i < decrement; i++) {
managedBuffer.release();
}
}
return super.release(decrement);
}
}
57 changes: 57 additions & 0 deletions common/network-common/src/main/java/org/apache/spark/network/sasl/SaslEncryption.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,63 @@ public long transfered() {
return transferred;
}

@Override
public long transferred() {
return transferred;
}

/**
* Override this due to different return types of ReferenceCounted.touch and FileRegion.touch.
*/
@Override
public EncryptedMessage touch() {
super.touch();
return this;
}

@Override
public EncryptedMessage touch(Object o) {
if (buf != null) {
buf.touch(o);
}
if (region != null) {
region.touch(o);
}
return this;
}

/**
* Override this due to different return types of ReferenceCounted.retain and FileRegion.retain.
*/
@Override
public EncryptedMessage retain() {
super.retain();
return this;
}

@Override
public EncryptedMessage retain(int increment) {
super.retain(increment);
if (buf != null) {
buf.retain(increment);
}
if (region != null) {
region.retain(increment);
}
return this;
}

@Override
public boolean release(int decrement) {
if (region != null) {
region.release(decrement);
}
if (buf != null) {
buf.release(decrement);
}
return super.release(decrement);
}

/**
* Transfers data from the original message to the channel, encrypting it in the process.
*
Expand Down
4 changes: 2 additions & 2 deletions common/network-common/src/test/java/org/apache/spark/network/ProtocolSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@ private void testServerToClient(Message msg) {
NettyUtils.createFrameDecoder(), MessageDecoder.INSTANCE);

while (!serverChannel.outboundMessages().isEmpty()) {
clientChannel.writeInbound(serverChannel.readOutbound());
clientChannel.writeOneInbound(serverChannel.readOutbound());
}

assertEquals(1, clientChannel.inboundMessages().size());
Expand All @@ -72,7 +72,7 @@ private void testClientToServer(Message msg) {
NettyUtils.createFrameDecoder(), MessageDecoder.INSTANCE);

while (!clientChannel.outboundMessages().isEmpty()) {
serverChannel.writeInbound(clientChannel.readOutbound());
serverChannel.writeOneInbound(clientChannel.readOutbound());
}

assertEquals(1, serverChannel.inboundMessages().size());
Expand Down
28 changes: 28 additions & 0 deletions ...etwork-common/src/test/java/org/apache/spark/network/protocol/MessageWithHeaderSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,34 @@ public long transfered() {
return 8 * written;
}

@Override
public long transferred() {
return 8 * written;
}

@Override
public TestFileRegion touch() {
super.touch();
return this;
}

@Override
public TestFileRegion touch(Object o) {
return this;
}

@Override
public TestFileRegion retain() {
super.retain();
return this;
}

@Override
public TestFileRegion retain(int increment) {
super.retain(increment);
return this;
}

@Override
public long transferTo(WritableByteChannel target, long position) throws IOException {
for (int i = 0; i < writesPerCall; i++) {
Expand Down
29 changes: 29 additions & 0 deletions core/src/main/scala/org/apache/spark/storage/DiskStore.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -279,6 +279,35 @@ private class ReadableChannelFileRegion(source: ReadableByteChannel, blockSize:

override def transfered(): Long = _transferred

override def transferred(): Long = _transferred

/**
* Override this due to different return types of ReferenceCounted.touch and FileRegion.touch.
*/
override def touch(): this.type = {
super.touch()
this
}

override def touch(o: Object): this.type = {
this
}

/**
* Override this due to different return types of ReferenceCounted.retain and FileRegion.retain.
*/
override def retain(): this.type = {
super.retain()
this
}

override def retain(increment: Int): this.type = {
super.retain(increment)
this
}

override def release(decrement: Int): Boolean = super.release(decrement)

override def transferTo(target: WritableByteChannel, pos: Long): Long = {
assert(pos == transfered(), "Invalid position.")

Expand Down
2 changes: 1 addition & 1 deletion dev/deps/spark-deps-hadoop-2.6
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ metrics-jvm-3.1.5.jar
minlog-1.3.0.jar
mx4j-3.0.2.jar
netty-3.9.9.Final.jar
netty-all-4.0.47.Final.jar
netty-all-4.1.17.Final.jar
objenesis-2.1.jar
opencsv-2.3.jar
orc-core-1.4.1-nohive.jar
Expand Down
2 changes: 1 addition & 1 deletion dev/deps/spark-deps-hadoop-2.7
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,7 +147,7 @@ metrics-jvm-3.1.5.jar
minlog-1.3.0.jar
mx4j-3.0.2.jar
netty-3.9.9.Final.jar
netty-all-4.0.47.Final.jar
netty-all-4.1.17.Final.jar
objenesis-2.1.jar
opencsv-2.3.jar
orc-core-1.4.1-nohive.jar
Expand Down
2 changes: 1 addition & 1 deletion pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -580,7 +580,7 @@
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.0.47.Final</version>
<version>4.1.17.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
Expand Down