spark源码阅读之network(1)

spark将在1.6中替换掉akka，而采用netty实现整个集群的rpc的框架，netty的内存管理和NIO支持将有效的提高spark集群的网络传输能力，为了看懂这块代码，在网上找了两本书看《netty in action》和《netty权威指南》，结合了spark的源码既学习了netty也看完了spark netty的部分源码。该部分源码掺杂了太多netty的东西，看起来还是有点累的。

下面是我画的UML类图。https://onedrive.live.com/redir?resid=C6D5F843CD40C4E8!364&authkey=!ABHmp7kWEm5n9Y4&ithint=file%2cEAP

 

缓存模块

network工程里面抽闲了一个ManagerBuffer的接口，该接口用来表示二进制数据中视图（表示数据的一部分），具体的实现依赖数据的来源，目前支持file,nio bytebuffer,netty bytebuf这3中数据来源。注意该接口具体的实现可能脱离了JVM GC的管理，比如NettyManagerBuffer是引用计数的，此时当该buffer传递给其他线程是需要调用retain/release来添加或减少引用。

ManagerBuffer以ByteBuffer, InputStream和Netty对象三种方式对外显示这些数据，ByteBuffer由于消耗过大，不建议使用，添加了引用计数管理和数据大小查询。

1. publicabstractclassManagedBuffer{
3. /** Number of bytes of the data. */
4. publicabstractlong size();
6. /**
7. * Exposes this buffer's data as an NIO ByteBuffer. Changing the position and limit of the
8. * returned ByteBuffer should not affect the content of this buffer.
9. */
10. // TODO: Deprecate this, usage may require expensive memory mapping or allocation.
11. publicabstractByteBuffer nioByteBuffer()throwsIOException;
13. /**
14. * Exposes this buffer's data as an InputStream. The underlying implementation does not
15. * necessarily check for the length of bytes read, so the caller is responsible for making sure
16. * it does not go over the limit.
17. */
18. publicabstractInputStream createInputStream()throwsIOException;
20. /**
21. * Increment the reference count by one if applicable.
22. */
23. publicabstractManagedBuffer retain();
25. /**
26. * If applicable, decrement the reference count by one and deallocates the buffer if the
27. * reference count reaches zero.
28. */
29. publicabstractManagedBuffer release();
31. /**
32. * Convert the buffer into an Netty object, used to write the data out.
33. */
34. publicabstractObject convertToNetty()throwsIOException;
35. }

ManageredBuffer每一种数据来源有一个实现类。先看下数据来源为file的。

1. publicfinalclassFileSegmentManagedBufferextendsManagedBuffer{
2. privatefinalTransportConf conf;
3. privatefinalFile file;
4. privatefinallong offset;
5. privatefinallong length;
7. publicFileSegmentManagedBuffer(TransportConf conf,File file,long offset,long length){
8. this.conf = conf;
9. this.file = file;
10. this.offset = offset;
11. this.length = length;
12. }
14. @Override
15. publiclong size(){
16. return length;
17. }
19. @Override
20. publicByteBuffer nioByteBuffer()throwsIOException{
21. FileChannel channel =null;
22. try{
23. channel =newRandomAccessFile(file,"r").getChannel();
24. // Just copy the buffer if it's sufficiently small, as memory mapping has a high overhead.
25. if(length < conf.memoryMapBytes()){
26. ByteBuffer buf =ByteBuffer.allocate((int) length);
27. channel.position(offset);
28. while(buf.remaining()!=0){
29. if(channel.read(buf)==-1){
30. thrownewIOException(String.format("Reached EOF before filling buffer "+
31. "offset=%s file=%s buf.remaining=%s",
32. offset, file.getAbsoluteFile(), buf.remaining()));
33. }
34. }
35. buf.flip();
36. return buf;
37. }else{
38. return channel.map(FileChannel.MapMode.READ_ONLY, offset, length);
39. }
40. }catch(IOException e){
41. try{
42. if(channel !=null){
43. long size = channel.size();
44. thrownewIOException("Error in reading "+this+" (actual file length "+ size +")",
45. e);
46. }
47. }catch(IOException ignored){
48. // ignore
49. }
50. thrownewIOException("Error in opening "+this, e);
51. }finally{
52. JavaUtils.closeQuietly(channel);
53. }
54. }
56. @Override
57. publicInputStream createInputStream()throwsIOException{
58. FileInputStream is =null;
59. try{
60. is =newFileInputStream(file);
61. ByteStreams.skipFully(is, offset);
62. returnnewLimitedInputStream(is, length);
63. }catch(IOException e){
64. try{
65. if(is !=null){
66. long size = file.length();
67. thrownewIOException("Error in reading "+this+" (actual file length "+ size +")",
68. e);
69. }
70. }catch(IOException ignored){
71. // ignore
72. }finally{
73. JavaUtils.closeQuietly(is);
74. }
75. thrownewIOException("Error in opening "+this, e);
76. }catch(RuntimeException e){
77. JavaUtils.closeQuietly(is);
78. throw e;
79. }
80. }
82. @Override
83. publicManagedBuffer retain(){
84. returnthis;
85. }
87. @Override
88. publicManagedBuffer release(){
89. returnthis;
90. }
92. @Override
93. publicObject convertToNetty()throwsIOException{
94. if(conf.lazyFileDescriptor()){
95. returnnewLazyFileRegion(file, offset, length);
96. }else{
97. FileChannel fileChannel =newFileInputStream(file).getChannel();
98. returnnewDefaultFileRegion(fileChannel, offset, length);
99. }
100. }
102. publicFile getFile(){return file;}
104. publiclong getOffset(){return offset;}
106. publiclong getLength(){return length;}
108. @Override
109. publicString toString(){
110. returnObjects.toStringHelper(this)
111. .add("file", file)
112. .add("offset", offset)
113. .add("length", length)
114. .toString();
115. }
116. }

nioByteBuffer,如果数据大小小于spark.storage.memoryMapThreshold。那么使用ByteBufer读取通道的数据，如果大于等于该值，那么使用文件内存映射方式读取数据。

createInputStream中返回一个控制读取长度的LimitedInputStream，这里使用guava的ByteStreams

convertToNetty返回一个FileRegion。如果spark.shuffle.io.lazyFD设置为true那么使用LazyFileRegion，如果为false使用DefaultFileRegion。LazyFileRegion会在传输的时候生成FileChannel,注解说如果netty使用了epoll协议那么不可以使用LazyFileRegion。

 

数据源为ByteBuf的实现类，该类用Bytebuf来存储数据。

1. publicfinalclassNettyManagedBufferextendsManagedBuffer{
2. privatefinalByteBuf buf;
4. publicNettyManagedBuffer(ByteBuf buf){
5. this.buf = buf;
6. }
8. @Override
9. publiclong size(){
10. return buf.readableBytes();
11. }
13. @Override
14. publicByteBuffer nioByteBuffer()throwsIOException{
15. return buf.nioBuffer();
16. }
18. @Override
19. publicInputStream createInputStream()throwsIOException{
20. returnnewByteBufInputStream(buf);
21. }
23. @Override
24. publicManagedBuffer retain(){
25. buf.retain();
26. returnthis;
27. }
29. @Override
30. publicManagedBuffer release(){
31. buf.release();
32. returnthis;
33. }
35. @Override
36. publicObject convertToNetty()throwsIOException{
37. return buf.duplicate();
38. }
40. @Override
41. publicString toString(){
42. returnObjects.toStringHelper(this)
43. .add("buf", buf)
44. .toString();
45. }
46. }

把一个bytebuf对象转成InputStream对象使用ByteBufInputStream对象来完成。还有bytebuf的duplicate()返回一个bytebuf映射同一份数据，任何一个修改结果都会影响另一个，注意引用计数。参见http://www.maljob.com/pages/newsDetail.html?id=394

 

还一个数据源为bytebuffer的实现

1. publicfinalclassNioManagedBufferextendsManagedBuffer{
2. privatefinalByteBuffer buf;
4. publicNioManagedBuffer(ByteBuffer buf){
5. this.buf = buf;
6. }
8. @Override
9. publiclong size(){
10. return buf.remaining();
11. }
13. @Override
14. publicByteBuffer nioByteBuffer()throwsIOException{
15. return buf.duplicate();
16. }
18. @Override
19. publicInputStream createInputStream()throwsIOException{
20. returnnewByteBufInputStream(Unpooled.wrappedBuffer(buf));
21. }
23. @Override
24. publicManagedBuffer retain(){
25. returnthis;
26. }
28. @Override
29. publicManagedBuffer release(){
30. returnthis;
31. }
33. @Override
34. publicObject convertToNetty()throwsIOException{
35. returnUnpooled.wrappedBuffer(buf);
36. }
38. @Override
39. publicString toString(){
40. returnObjects.toStringHelper(this)
41. .add("buf", buf)
42. .toString();
43. }
44. }

 这里面一个有意思的显示就是把bytebuffer转成bytebuf使用netty中Unpooled.wrappedBuffer()实现

 

 

 

 

来自为知笔记(Wiz)