{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/14995","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/14995/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/14995/comments","events_url":"https://api.github.com/repos/apache/spark/issues/14995/events","html_url":"https://github.com/apache/spark/pull/14995","id":175484317,"number":14995,"title":"[Test Only][not ready for review][SPARK-6235][CORE]Address various 2G limits ","user":{"login":"witgo","id":302879,"avatar_url":"https://avatars.githubusercontent.com/u/302879?v=3","gravatar_id":"","url":"https://api.github.com/users/witgo","html_url":"https://github.com/witgo","followers_url":"https://api.github.com/users/witgo/followers","following_url":"https://api.github.com/users/witgo/following{/other_user}","gists_url":"https://api.github.com/users/witgo/gists{/gist_id}","starred_url":"https://api.github.com/users/witgo/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/witgo/subscriptions","organizations_url":"https://api.github.com/users/witgo/orgs","repos_url":"https://api.github.com/users/witgo/repos","events_url":"https://api.github.com/users/witgo/events{/privacy}","received_events_url":"https://api.github.com/users/witgo/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":3,"created_at":"2016-09-07T11:59:32Z","updated_at":"2016-09-07T12:08:39Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/14995","html_url":"https://github.com/apache/spark/pull/14995","diff_url":"https://github.com/apache/spark/pull/14995.diff","patch_url":"https://github.com/apache/spark/pull/14995.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\n### motivation\r\n\r\nThe various 2G limit in Spark. \r\n\r\n1. When reading the data block is stored in the hard disk, the following code fragment is called. \r\n\r\n  ```scala\r\n    val iterToReturn: Iterator[Any] = {\r\n      val diskBytes = diskStore.getBytes(blockId)\r\n      if (level.deserialized) {\r\n        val diskValues = serializerManager.dataDeserializeStream(\r\n          blockId,\r\n          diskBytes.toInputStream(dispose = true))(info.classTag)\r\n        maybeCacheDiskValuesInMemory(info, blockId, level, diskValues)\r\n      } else {\r\n        val stream = maybeCacheDiskBytesInMemory(info, blockId, level, diskBytes)\r\n          .map {_.toInputStream(dispose = false)}\r\n          .getOrElse { diskBytes.toInputStream(dispose = true) }\r\n        serializerManager.dataDeserializeStream(blockId, stream)(info.classTag)\r\n      }\r\n    }\r\n\r\n  ```\r\n\r\n  ```scala\r\n    def getBytes(blockId: BlockId): ChunkedByteBuffer = {\r\n      val file = diskManager.getFile(blockId.name)\r\n      val channel = new RandomAccessFile(file, \"r\").getChannel\r\n      Utils.tryWithSafeFinally {\r\n        // For small files, directly read rather than memory map\r\n        if (file.length < minMemoryMapBytes) {\r\n          val buf = ByteBuffer.allocate(file.length.toInt)\r\n          channel.position(0)\r\n          while (buf.remaining() != 0) {\r\n            if (channel.read(buf) == -1) {\r\n              throw new IOException(\"Reached EOF before filling buffer\\n\" +\r\n                s\"offset=0\\nfile=${file.getAbsolutePath}\\nbuf.remaining=${buf.remaining}\")\r\n            }\r\n          }\r\n          buf.flip()\r\n          new ChunkedByteBuffer(buf)\r\n        } else {\r\n          new ChunkedByteBuffer(channel.map(MapMode.READ_ONLY, 0, file.length))\r\n        }\r\n      } {\r\n        channel.close()\r\n      }\r\n    }\r\n\r\n  ```\r\n\r\n  The above code has the following problems:  \r\n    * `channel.map(MapMode.READ_ONLY, 0, file.length)` returns an instance of `MappedByteBuffer`. the size of `MappedByteBuffer` can not exceed 2G. \r\n    * When a `Iterator[Any]` is generated, need to load all the data into the memory,this may take up a lot of memory.\r\n\r\n\r\n2. When using kryo serialized data, the following code fragment is called: \r\n\r\n  ```scala\r\n\r\n    override def serialize[T: ClassTag](t: T): ByteBuffer = {\r\n      output.clear()\r\n      val kryo = borrowKryo()\r\n      try {\r\n        kryo.writeClassAndObject(output, t)\r\n      } catch {\r\n        case e: KryoException if e.getMessage.startsWith(\"Buffer overflow\") =>\r\n          throw new SparkException(s\"Kryo serialization failed: ${e.getMessage}. To avoid this, \" +\r\n            \"increase spark.kryoserializer.buffer.max value.\")\r\n      } finally {\r\n        releaseKryo(kryo)\r\n      }\r\n      ByteBuffer.wrap(output.toBytes)\r\n    }\r\n\r\n  ```\r\n\r\n  The above code has the following problems:\r\n    * The serialization data is stored in the `output` internal `byte[]`, the size of `byte[]` can not exceed 2G. \r\n3. When RPC writes data to be sent to the Channel, the following code fragment is called: \r\n  ```scala\r\n    public long transferTo(final WritableByteChannel target, final long position) throws IOException {\r\n      Preconditions.checkArgument(position == totalBytesTransferred, \"Invalid position.\");\r\n      // Bytes written for header in this call.\r\n      long writtenHeader = 0;\r\n      if (header.readableBytes() > 0) {\r\n        writtenHeader = copyByteBuf(header, target);\r\n        totalBytesTransferred += writtenHeader;\r\n        if (header.readableBytes() > 0) {\r\n          return writtenHeader;\r\n        }\r\n      }\r\n\r\n      // Bytes written for body in this call.\r\n      long writtenBody = 0;\r\n      if (body instanceof FileRegion) {\r\n        writtenBody = ((FileRegion) body).transferTo(target, totalBytesTransferred - headerLength);\r\n      } else if (body instanceof ByteBuf) {\r\n        writtenBody = copyByteBuf((ByteBuf) body, target);\r\n      }\r\n      totalBytesTransferred += writtenBody;\r\n      return writtenHeader + writtenBody;\r\n    }\r\n  ```\r\n\r\n  The above code has the following problems: \r\n    * the size of ByteBuf cannot exceed 2G \r\n    * cannot transfer data over 2G in memory\r\n\r\n    \r\n4. When decodes the RPC message received, the following code fragment is called: \r\n\r\n  ```scala\r\n  public final class MessageDecoder extends MessageToMessageDecoder<ByteBuf> {\r\n\r\n    private static final Logger logger = LoggerFactory.getLogger(MessageDecoder.class);\r\n\r\n    @Override\r\n    public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {\r\n      Message.Type msgType = Message.Type.decode(in);\r\n      Message decoded = decode(msgType, in);\r\n      assert decoded.type() == msgType;\r\n      logger.trace(\"Received message {}: {}\", msgType, decoded);\r\n      out.add(decoded);\r\n    }\r\n\r\n    private Message decode(Message.Type msgType, ByteBuf in) {\r\n      switch (msgType) {\r\n        case ChunkFetchRequest:\r\n          return ChunkFetchRequest.decode(in);\r\n\r\n        case ChunkFetchSuccess:\r\n          return ChunkFetchSuccess.decode(in);\r\n\r\n        case ChunkFetchFailure:\r\n          return ChunkFetchFailure.decode(in);\r\n\r\n        default:\r\n          throw new IllegalArgumentException(\"Unexpected message type: \" + msgType);\r\n      }\r\n    }\r\n  }\r\n\r\n  ```\r\n\r\n  The above code has the following problems: \r\n    * the size of ByteBuf cannot exceed 2G \r\n    * Must be in the receiver to complete the data can be decoded\r\n\r\n### Goals\r\n  \r\n* Setup for eliminating the various 2G limit in Spark.   (The 2G limit 1,2,3,4)\r\n* Support back-pressure flow control for remote data reading(experimental goal).  (The 2G limit 4)\r\n* Add buffer pool(long-range goal). \r\n\r\n### Design\r\n\r\n#### Setup for eliminating the various 2G limit in Spark.  \r\n\r\n##### Replace ByteBuffer with ChunkedByteBuffer. (The 2G limit 1,2)\r\n  * Support reference counting, a necessary condition to the feature of the buffer pool \r\n      [Reference counted objects](http://netty.io/wiki/reference-counted-objects.html)\r\n  * Support serialization for easy transport \r\n  * Support slice duplicate and copy operation \r\n  * Can be efficiently converted to `InputStream`, `ByteBuffer`, `byte[]` and `ByteBuf`, etc. \r\n\r\n1. Move the ChunkedByteBuffer class to `common/network-common/src/main/java/org/apache/spark/network/buffer/`. \r\n2. Modify `ManagedBuffer.nioByteBuffer`'s return value to ChunkedByteBuffer instance.(The 2G limit 1)\r\n3. Modify the parameter of `SerializerInstance.deserialize` and the return value of `SerializerInstance.serialize` to ChunkedByteBuffer instance. (The 2G limit 2)\r\n\r\n  ```scala\r\n  def serialize[T: ClassTag](t: T): ChunkedByteBuffer = {\r\n    output.clear()\r\n    val out = ChunkedByteBufferOutputStream.newInstance()\r\n    output.setOutputStream(out)\r\n    val kryo = borrowKryo()\r\n    kryo.writeClassAndObject(output, t)\r\n    output.close()\r\n    out.toChunkedByteBuffer\r\n  }\r\n  ```\r\n\r\n4. Other changes. \r\n\r\n##### Replace ByteBuf with InputStream. \r\n\r\n1. Modify `NioManagedBuffer.convertToNetty` method returns InputStream instances when data is larger than Integer.MAX_VALUE. (The 2G limit 3)\r\n2. Add InputStreamManagedBuffer class, used to convert InputStream instance to ManagedBuffer instance. (The 2G limit 4)\r\n3. Modify MessageWithHeader classes, support processing InputStream instance (The 2G limit 3)\r\n4. Modify the parameters of the `Encodable.encode` method to OutputStream instance. (The 2G limit 4)\r\n5. Modify the parameters of the decode method of the classes who implement the Encodable interface to InputStream instance.It can handle mixed storage data  (The 2G limit 3) \r\n  \r\n  ```scala\r\n  public InputStream toInputStream() throws IOException {\r\n    ChunkedByteBufferOutputStream out = ChunkedByteBufferOutputStream.newInstance();\r\n    Encoders.Bytes.encode(out, type().id());\r\n    encodeWithoutBlockData(out);\r\n    // out.toChunkedByteBuffer().toInputStream() data in memory\r\n    // blockData.createInputStream()  data in hard disk(FileInputStream)\r\n    return new SequenceInputStream(out.toChunkedByteBuffer().toInputStream(),\r\n        blockData.createInputStream());\r\n  }\r\n  ```\r\n\r\n6. Modify TransportFrameDecoder class, use `LinkedList<ByteBuf>` to represent the Frame, remove the size limit of Frame.  The 2G limit 4) \r\n7. Add ByteBufInputStream class, used to convert `LinkedList<ByteBuf>` instance to InputStream instance.  (The 2G limit 4) \r\n8. Modify the parameters of `RpcHandler.receive` method to InputStream instance.  (The 2G limit 4) \r\n\r\n#### Read data\r\n\r\n##### Local data\r\n\r\n1. Only the data stored in the memory is represented by ChunkedByteBuffer, the other is represented by ManagedBuffer.  (The 2G limit 1) \r\n\r\n  * Modify `DiskStore.getBytes`'s return value type to  ManagedBuffer instance,  which calls `ManagedBuffer.nioByteBuffer` only when the memory has enough space to store the ManagedBuffer data.\r\n\r\n##### Remote Data (The 2G limit 4)\r\n\r\nThere are three options: \r\n\r\n1. Add InputStreamInterceptor to support propagate back-pressure to shuffle server(The option has been implemented): \r\n  * When the number of ByteBuf in the cache exceeds a certain amount, call `channel.config ().SetAutoRead (false) ` disable AUTO_READ, no longer automatically call `channle.read ()`. \r\n  * When the number of ByteBuf in the cache is smaller than a certain amount, call `channel.config().setAutoRead(true)` enable AUTO_READ . \r\n  * The advantage of this option is to support propagate back-pressure; drawback is that can lead semantic change the existing API, in some cases the IO retry function is invalid. \r\n\r\n2.  When the size of message is greater than a certain value, the message is written to disk, not take up memory. ma\r\n  * The advantage of this options is to take up very little memory, the disadvantage is to increase the disk IO. \r\n3. Combined with buffer pool, qs far as possible stores data in memory. \r\n  * Write message to the buffer pool when there has enough memory, otherwise write on disk. \r\n\r\n## How was this patch tested?\r\n\r\nTODO\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/245260300","html_url":"https://github.com/apache/spark/pull/14995#issuecomment-245260300","issue_url":"https://api.github.com/repos/apache/spark/issues/14995","id":245260300,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars.githubusercontent.com/u/4031291?v=3","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2016-09-07T12:08:39Z","updated_at":"2016-09-07T12:08:39Z","body":"Merged build finished. Test FAILed."}}
{"actor":{"display_login":"AmplabJenkins"}}