Rename limit to size.

core/src/main/scala/org/apache/spark/storage/BlockManager.scala

@@ -503,7 +503,7 @@ private[spark] class BlockManager(
    */
   def getRemoteValues(blockId: BlockId): Option[BlockResult] = {
     getRemoteBytes(blockId).map { data =>
-      new BlockResult(dataDeserialize(blockId, data), DataReadMethod.Network, data.limit)
+      new BlockResult(dataDeserialize(blockId, data), DataReadMethod.Network, data.size)
     }
   }
 
@@ -742,7 +742,7 @@ private[spark] class BlockManager(
         null
       }
 
-      val size = bytes.limit
+      val size = bytes.size
 
       if (level.useMemory) {
         // Put it in memory first, even if it also has useDisk set to true;
@@ -950,7 +950,7 @@ private[spark] class BlockManager(
           diskBytes.dispose()
           memoryStore.getBytes(blockId).get
         } else {
-          val putSucceeded = memoryStore.putBytes(blockId, diskBytes.limit, () => {
+          val putSucceeded = memoryStore.putBytes(blockId, diskBytes.size, () => {
             // https://issues.apache.org/jira/browse/SPARK-6076
             // If the file size is bigger than the free memory, OOM will happen. So if we
             // cannot put it into MemoryStore, copyForMemory should not be created. That's why
@@ -1079,15 +1079,15 @@ private[spark] class BlockManager(
         case Some(peer) =>
           try {
             val onePeerStartTime = System.currentTimeMillis
-            logTrace(s"Trying to replicate $blockId of ${data.limit} bytes to $peer")
+            logTrace(s"Trying to replicate $blockId of ${data.size} bytes to $peer")
             blockTransferService.uploadBlockSync(
               peer.host,
               peer.port,
               peer.executorId,
               blockId,
               new NettyManagedBuffer(data.toNetty),
               tLevel)
-            logTrace(s"Replicated $blockId of ${data.limit} bytes to $peer in %s ms"
+            logTrace(s"Replicated $blockId of ${data.size} bytes to $peer in %s ms"
               .format(System.currentTimeMillis - onePeerStartTime))
             peersReplicatedTo += peer
             peersForReplication -= peer
@@ -1110,7 +1110,7 @@ private[spark] class BlockManager(
       }
     }
     val timeTakeMs = (System.currentTimeMillis - startTime)
-    logDebug(s"Replicating $blockId of ${data.limit} bytes to " +
+    logDebug(s"Replicating $blockId of ${data.size} bytes to " +
       s"${peersReplicatedTo.size} peer(s) took $timeTakeMs ms")
     if (peersReplicatedTo.size < numPeersToReplicateTo) {
       logWarning(s"Block $blockId replicated to only " +

core/src/main/scala/org/apache/spark/storage/memory/MemoryStore.scala

@@ -99,7 +99,7 @@ private[spark] class MemoryStore(
     if (memoryManager.acquireStorageMemory(blockId, size)) {
       // We acquired enough memory for the block, so go ahead and put it
       val bytes = _bytes()
-      assert(bytes.limit == size)
+      assert(bytes.size == size)
       val entry = new SerializedMemoryEntry(bytes, size)
       entries.synchronized {
         entries.put(blockId, entry)
@@ -189,7 +189,7 @@ private[spark] class MemoryStore(
         new DeserializedMemoryEntry(arrayValues, SizeEstimator.estimate(arrayValues))
       } else {
         val bytes = blockManager.dataSerialize(blockId, arrayValues.iterator)
-        new SerializedMemoryEntry(bytes, bytes.limit)
+        new SerializedMemoryEntry(bytes, bytes.size)
       }
       val size = entry.size
       def transferUnrollToStorage(amount: Long): Unit = {

core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala

@@ -27,12 +27,13 @@ import io.netty.buffer.{ByteBuf, Unpooled}
 import org.apache.spark.network.util.ByteArrayWritableChannel
 import org.apache.spark.storage.StorageUtils
 
+
 private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
   require(chunks != null, "chunks must not be null")
   require(chunks.forall(_.limit() > 0), "chunks must be non-empty")
   require(chunks.forall(_.position() == 0), "chunks' positions must be 0")
 
-  val limit: Long = chunks.map(_.limit().asInstanceOf[Long]).sum
+  val size: Long = chunks.map(_.limit().asInstanceOf[Long]).sum
 
   def this(byteBuffer: ByteBuffer) = {
     this(Array(byteBuffer))
@@ -56,11 +57,11 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
    * @throws UnsupportedOperationException if this buffer's size exceeds the maximum array size.
    */
   def toArray: Array[Byte] = {
-    if (limit >= Integer.MAX_VALUE) {
+    if (size >= Integer.MAX_VALUE) {
       throw new UnsupportedOperationException(
-        s"cannot call toArray because buffer size ($limit bytes) exceeds maximum array size")
+        s"cannot call toArray because buffer size ($size bytes) exceeds maximum array size")
     }
-    val byteChannel = new ByteArrayWritableChannel(limit.toInt)
+    val byteChannel = new ByteArrayWritableChannel(size.toInt)
     writeFully(byteChannel)
     byteChannel.close()
     byteChannel.getData
@@ -81,6 +82,7 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
 
   /**
    * Creates an input stream to read data from this ChunkedByteBuffer.
+ *
    * @param dispose if true, [[dispose()]] will be called at the end of the stream
    *                in order to close any memory-mapped files which back this buffer.
    */

core/src/test/scala/org/apache/spark/io/ChunkedByteBufferSuite.scala

@@ -29,7 +29,7 @@ class ChunkedByteBufferSuite extends SparkFunSuite {
 
   test("no chunks") {
     val emptyChunkedByteBuffer = new ChunkedByteBuffer(Array.empty[ByteBuffer])
-    assert(emptyChunkedByteBuffer.limit === 0)
+    assert(emptyChunkedByteBuffer.size === 0)
     assert(emptyChunkedByteBuffer.getChunks().isEmpty)
     assert(emptyChunkedByteBuffer.toArray === Array.empty)
     assert(emptyChunkedByteBuffer.toByteBuffer.capacity() === 0)
@@ -58,7 +58,7 @@ class ChunkedByteBufferSuite extends SparkFunSuite {
 
   test("writeFully() does not affect original buffer's position") {
     val chunkedByteBuffer = new ChunkedByteBuffer(Array(ByteBuffer.allocate(8)))
-    chunkedByteBuffer.writeFully(new ByteArrayWritableChannel(chunkedByteBuffer.limit.toInt))
+    chunkedByteBuffer.writeFully(new ByteArrayWritableChannel(chunkedByteBuffer.size.toInt))
     assert(chunkedByteBuffer.getChunks().head.position() === 0)
   }
 
@@ -72,7 +72,7 @@ class ChunkedByteBufferSuite extends SparkFunSuite {
     val fourMegabyteBuffer = ByteBuffer.allocate(1024 * 1024 * 4)
     fourMegabyteBuffer.limit(fourMegabyteBuffer.capacity())
     val chunkedByteBuffer = new ChunkedByteBuffer(Array.fill(1024)(fourMegabyteBuffer))
-    assert(chunkedByteBuffer.limit === (1024L * 1024L * 1024L * 4L))
+    assert(chunkedByteBuffer.size === (1024L * 1024L * 1024L * 4L))
     intercept[UnsupportedOperationException] {
       chunkedByteBuffer.toArray
     }
@@ -82,10 +82,10 @@ class ChunkedByteBufferSuite extends SparkFunSuite {
     val bytes1 = ByteBuffer.wrap(Array.tabulate(256)(_.toByte))
     val bytes2 = ByteBuffer.wrap(Array.tabulate(128)(_.toByte))
     val chunkedByteBuffer = new ChunkedByteBuffer(Array(bytes1, bytes2))
-    assert(chunkedByteBuffer.limit === bytes1.limit() + bytes2.limit())
+    assert(chunkedByteBuffer.size === bytes1.limit() + bytes2.limit())
 
     val inputStream = chunkedByteBuffer.toInputStream(dispose = false)
-    val bytesFromStream = new Array[Byte](chunkedByteBuffer.limit.toInt)
+    val bytesFromStream = new Array[Byte](chunkedByteBuffer.size.toInt)
     ByteStreams.readFully(inputStream, bytesFromStream)
     assert(bytesFromStream === bytes1.array() ++ bytes2.array())
     assert(chunkedByteBuffer.getChunks().head.position() === 0)