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\n"+
31. "offset=%s\nfile=%s\nbuf.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)