Skip to content

[SPARK-21527][CORE] Use buffer limit in order to use JAVA NIO Util's buffercache #18730

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
caneGuy wants to merge 19 commits into from
Closed

[SPARK-21527][CORE] Use buffer limit in order to use JAVA NIO Util's buffercache #18730

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
7cbadc5
[SPARK][CORE] Slice write by channel
caneMi Jul 25, 2017
fc91f96
Fix code style
caneGuy Aug 23, 2017
bab91db
Refine to avoid underflow
caneGuy Aug 24, 2017
72aef67
Fix no ending loop
caneGuy Aug 24, 2017
4789772
Remove useless unit test
caneGuy Aug 24, 2017
aeabe1d
Delete unused package
caneGuy Aug 24, 2017
ca77b51
Add benchmark testing with Utils.BenchMark
caneGuy Aug 24, 2017
e84a6d7
Refine code style
caneGuy Aug 24, 2017
d708142
Update as comment replace writeFully with slice logic
caneGuy Aug 24, 2017
33a2796
Remove unused package
caneGuy Aug 24, 2017
ab384d4
No need call toInt
caneGuy Aug 24, 2017
b669351
Only need to use bytes.remaining()
caneGuy Aug 24, 2017
717f886
Use config for NIO_BUFFER_LIMIT
caneGuy Aug 24, 2017
9d3004d
Use BUFFER_WRITE_CHUNK_SIZE
caneGuy Aug 24, 2017
e48f44f
Update as commented
caneGuy Aug 24, 2017
fc184aa
Fix build error
caneGuy Aug 24, 2017
f1d67a3
Fix compile error and refine scala style
caneGuy Aug 24, 2017
14ca824
Need to convert to int to fix compile error
caneGuy Aug 25, 2017
a02575e
Avoid per-loop type cast
caneGuy Aug 25, 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
2 changes: 1 addition & 1 deletion 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 @@ -93,7 +93,7 @@ private[spark] class DiskStore(

def putBytes(blockId: BlockId, bytes: ChunkedByteBuffer): Unit = {
put(blockId) { channel =>
bytes.writeFully(channel)
bytes.writeWithSlice(channel)
}
}

Expand Down
15 changes: 15 additions & 0 deletions core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
require(chunks != null, "chunks must not be null")
require(chunks.forall(_.position() == 0), "chunks' positions must be 0")

private val NIO_BUFFER_LIMIT = 64 * 1024 * 1024 // Chunk size in bytes
Copy link
Contributor

@cloud-fan cloud-fan Aug 24, 2017

Choose a reason for hiding this comment

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

have you tested with different numbers? is 64mb the best choice?

Copy link
Contributor Author

@caneGuy caneGuy Aug 24, 2017

Choose a reason for hiding this comment

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

actually i did not tested too much, it depends on value from our cluster.And i do some simple test then found 64mb will not affect perfomance compared with logic before.And benifit is we only need one buffer cache for block larger than 64 mb

Copy link
Contributor

@cloud-fan cloud-fan Aug 24, 2017

Choose a reason for hiding this comment

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

shall we make it configurable? we can add a new entry in org.apache.spark.internal.config, and mark it as an internal config first.

Copy link
Contributor Author

@caneGuy caneGuy Aug 24, 2017

Choose a reason for hiding this comment

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

Ok, i will refactor later.Thanks for your suggestion.


private[this] var disposed: Boolean = false

/**
Expand All @@ -62,6 +64,19 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}

/**
* Write this buffer to a channel with slice.
*/
def writeWithSlice(channel: WritableByteChannel): Unit = {
Copy link
Contributor

@cloud-fan cloud-fan Aug 23, 2017

Choose a reason for hiding this comment

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

can we use this one to replace writeFully?

for (bytes <- getChunks()) {
val capacity = bytes.limit()
Copy link
Contributor

@cloud-fan cloud-fan Aug 24, 2017

Choose a reason for hiding this comment

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

this line is not used.

while (bytes.position() < capacity) {
bytes.limit(Math.min(capacity, bytes.position + NIO_BUFFER_LIMIT.toInt))
Copy link
Member

@kiszk kiszk Aug 23, 2017 •
edited
Loading

Choose a reason for hiding this comment

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

Should we replace Math.min(...) with Math.min(capacity, bytes.position + NIO_BUFFER_LIMIT.toLong)? I am afraid about int underflow. For example, if capacity = 0x7FFFFFF0 and bytes.position = 0x7FFFFF00, the result of bytes.position + NIO_BUFFER_LIMIT.toInt is negative (i.e. greater than 0x80000000).
To avoid this underflow, it would be good to compare them by using long.

Copy link
Contributor Author

@caneGuy caneGuy Aug 24, 2017

Choose a reason for hiding this comment

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

Good review.I refactor the code.

channel.write(bytes)
}
}
}

/**
* Wrap this buffer to view it as a Netty ByteBuf.
*/
Expand Down