spark源码阅读之network(1)

时间：2024-08-06 12:06:08

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

缓存模块

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()实现

相关文章