You are viewing a plain text version of this content. The canonical link for it is here.

Posted to reviews@spark.apache.org by zhzhan <gi...@git.apache.org> on 2015/04/09 01:33:51 UTC

[GitHub] spark pull request: [SPARK-6479][SPARK-CORE]Create off-heap block ...

GitHub user zhzhan opened a pull request:

    https://github.com/apache/spark/pull/5430

    [SPARK-6479][SPARK-CORE]Create off-heap block storage API

    This is the classes for creating off-heap block storage API. Currently they are just stub classes, and the real migration for tachyon and new implementation for hdfs will be submit for review in spark-6112. 

You can merge this pull request into a Git repository by running:

    $ git pull https://github.com/zhzhan/spark SPARK-6479

Alternatively you can review and apply these changes as the patch at:

    https://github.com/apache/spark/pull/5430.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

    This closes #5430
    
----
commit b754d2adde8667473433cbf9397ae60cda170555
Author: Zhan Zhang <zh...@gmail.com>
Date:   2015-04-08T23:30:01Z

    initial commit for off-heap block storage api

----


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request: [SPARK-6479][Block Manager]Create off-heap blo...

Posted by zhzhan <gi...@git.apache.org>.

Github user zhzhan commented on the pull request:

    https://github.com/apache/spark/pull/5430#issuecomment-96123516
  
    Jenkins, retest this please.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request: [SPARK-6479][Block Manager]Create off-heap blo...

Posted by rxin <gi...@git.apache.org>.

Github user rxin commented on a diff in the pull request:

    https://github.com/apache/spark/pull/5430#discussion_r28178798
  
    --- Diff: core/src/main/scala/org/apache/spark/storage/OffHeapBlockManager.scala ---
    @@ -0,0 +1,109 @@
    +/*
    + * Licensed to the Apache Software Foundation (ASF) under one or more
    + * contributor license agreements.  See the NOTICE file distributed with
    + * this work for additional information regarding copyright ownership.
    + * The ASF licenses this file to You under the Apache License, Version 2.0
    + * (the "License"); you may not use this file except in compliance with
    + * the License.  You may obtain a copy of the License at
    + *
    + *    http://www.apache.org/licenses/LICENSE-2.0
    + *
    + * Unless required by applicable law or agreed to in writing, software
    + * distributed under the License is distributed on an "AS IS" BASIS,
    + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    + * See the License for the specific language governing permissions and
    + * limitations under the License.
    + */
    +
    +package org.apache.spark.storage
    +
    +import java.nio.ByteBuffer
    +import org.apache.spark.Logging
    +
    +import scala.util.control.NonFatal
    +
    +
    +trait OffHeapBlockManager {
    +
    +  /**
    +   * desc for the implementation.
    +   *
    +   */
    +  def desc(): String = {"OffHeap"}
    +
    +  /**
    +   * initialize a concrete block manager implementation.
    +   *
    +   * @throws java.io.IOException when FS init failure.
    +   */
    +  def init(blockManager: BlockManager, executorId: String): Unit
    +
    +  /**
    +   * remove the cache from offheap
    +   *
    +   * @throws java.io.IOException when FS failure in removing file.
    +   */
    +  def removeFile(blockId: BlockId): Boolean
    +
    +  /**
    +   * check the existence of the block cache
    +   *
    +   * @throws java.io.IOException when FS failure in checking the block existence.
    +   */
    +  def fileExists(blockId: BlockId): Boolean
    +
    +  /**
    +   * save the cache to the offheap.
    +   *
    +   * @throws java.io.IOException when FS failure in put blocks.
    +   */
    +  def putBytes(blockId: BlockId, bytes: ByteBuffer)
    +
    +  /**
    +   * retrieve the cache from offheap
    +   *
    +   * @throws java.io.IOException when FS failure in get blocks.
    +   */
    +  def getBytes(blockId: BlockId): Option[ByteBuffer]
    +
    +  /**
    +   * retrieve the size of the cache
    +   *
    +   * @throws java.io.IOException when FS failure in get block size.
    +   */
    +  def getSize(blockId: BlockId): Long
    +
    +  /**
    +   * cleanup when shutdown
    +   *
    +   */
    +  def addShutdownHook()
    +
    +  final def setup(blockManager: BlockManager, executorId: String): Unit = {
    +    init(blockManager, executorId)
    +    addShutdownHook()
    +  }
    +}
    +
    +object OffHeapBlockManager extends Logging{
    +  val MAX_DIR_CREATION_ATTEMPTS = 10
    +  val SUB_DIRS_PER_DIR = "64"
    +  def create(blockManager: BlockManager, executorId: String): Option[OffHeapBlockManager] = {
    --- End diff --
    
    add a blank line before def create
    
    we should add a default value so existing apps don't break.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request: [SPARK-6479][Block Manager]Create off-heap blo...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the pull request:

    https://github.com/apache/spark/pull/5430#issuecomment-97244316
  
    Merged build finished. Test FAILed.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request: [SPARK-6479][Block Manager]Create off-heap blo...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the pull request:

    https://github.com/apache/spark/pull/5430#issuecomment-97273537
  
     Merged build triggered.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request: [SPARK-6479][Block Manager]Create off-heap blo...

Posted by rxin <gi...@git.apache.org>.

Github user rxin commented on a diff in the pull request:

    https://github.com/apache/spark/pull/5430#discussion_r28927891
  
    --- Diff: core/src/main/scala/org/apache/spark/storage/ExtBlockManager.scala ---
    @@ -0,0 +1,81 @@
    +/*
    + * Licensed to the Apache Software Foundation (ASF) under one or more
    + * contributor license agreements.  See the NOTICE file distributed with
    + * this work for additional information regarding copyright ownership.
    + * The ASF licenses this file to You under the Apache License, Version 2.0
    + * (the "License"); you may not use this file except in compliance with
    + * the License.  You may obtain a copy of the License at
    + *
    + *    http://www.apache.org/licenses/LICENSE-2.0
    + *
    + * Unless required by applicable law or agreed to in writing, software
    + * distributed under the License is distributed on an "AS IS" BASIS,
    + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    + * See the License for the specific language governing permissions and
    + * limitations under the License.
    + */
    +
    +package org.apache.spark.storage
    +
    +import java.nio.ByteBuffer
    +import org.apache.spark.Logging
    +
    +import scala.util.control.NonFatal
    +
    +
    +private[spark] abstract class ExtBlockManager {
    +
    +  /**
    +   * desc for the implementation.
    --- End diff --
    
    Can you spend some time to improve the documentation of this API? Right now the javadocs are very sloppy. At the very least, we should:
    
    1. Write complete English sentences and words. For example "desc" is not a proper word.
    2. capitalize the first letters



---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request: [SPARK-6479][Block Manager]Create off-heap blo...

Posted by pwendell <gi...@git.apache.org>.

Github user pwendell commented on a diff in the pull request:

    https://github.com/apache/spark/pull/5430#discussion_r28916839
  
    --- Diff: core/src/main/scala/org/apache/spark/util/JsonProtocol.scala ---
    @@ -801,18 +801,18 @@ private[spark] object JsonProtocol {
       def storageLevelFromJson(json: JValue): StorageLevel = {
         val useDisk = (json \ "Use Disk").extract[Boolean]
         val useMemory = (json \ "Use Memory").extract[Boolean]
    -    val useTachyon = (json \ "Use Tachyon").extract[Boolean]
    +    val useExtBlkStore = (json \ "Use ExtBlkStore").extract[Boolean]
         val deserialized = (json \ "Deserialized").extract[Boolean]
         val replication = (json \ "Replication").extract[Int]
    -    StorageLevel(useDisk, useMemory, useTachyon, deserialized, replication)
    +    StorageLevel(useDisk, useMemory, useExtBlkStore, deserialized, replication)
       }
     
       def blockStatusFromJson(json: JValue): BlockStatus = {
         val storageLevel = storageLevelFromJson(json \ "Storage Level")
         val memorySize = (json \ "Memory Size").extract[Long]
         val diskSize = (json \ "Disk Size").extract[Long]
    -    val tachyonSize = (json \ "Tachyon Size").extract[Long]
    -    BlockStatus(storageLevel, memorySize, diskSize, tachyonSize)
    +    val extBlkStoreSize = (json \ "ExtBlkStore Size").extract[Long]
    --- End diff --
    
    If the "Tachyon" version is present, I'd look for that and then just convert it.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request: [SPARK-6479][Block Manager]Create off-heap blo...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the pull request:

    https://github.com/apache/spark/pull/5430#issuecomment-92549215
  
      [Test build #30203 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/30203/consoleFull) for   PR 5430 at commit [`0a49021`](https://github.com/apache/spark/commit/0a490215d1104ff37d2cf8971a78ac040b6875be).
     * This patch **passes all tests**.
     * This patch merges cleanly.
     * This patch adds no public classes.
     * This patch does not change any dependencies.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request: [SPARK-6479][Block Manager]Create off-heap blo...

Posted by rxin <gi...@git.apache.org>.

Github user rxin commented on a diff in the pull request:

    https://github.com/apache/spark/pull/5430#discussion_r28033633
  
    --- Diff: core/src/main/scala/org/apache/spark/storage/OffHeapStore.scala ---
    @@ -0,0 +1,142 @@
    +/*
    + * Licensed to the Apache Software Foundation (ASF) under one or more
    + * contributor license agreements.  See the NOTICE file distributed with
    + * this work for additional information regarding copyright ownership.
    + * The ASF licenses this file to You under the Apache License, Version 2.0
    + * (the "License"); you may not use this file except in compliance with
    + * the License.  You may obtain a copy of the License at
    + *
    + *    http://www.apache.org/licenses/LICENSE-2.0
    + *
    + * Unless required by applicable law or agreed to in writing, software
    + * distributed under the License is distributed on an "AS IS" BASIS,
    + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    + * See the License for the specific language governing permissions and
    + * limitations under the License.
    + */
    +
    +package org.apache.spark.storage
    +
    +import java.nio.ByteBuffer
    +import org.apache.spark.Logging
    +import org.apache.spark.util.Utils
    +
    +import scala.util.control.NonFatal
    +
    +
    +/**
    + * Stores BlockManager blocks on OffHeap.
    + * We capture any potential exception from underlying implementation
    + * and return with the expected failure value
    + */
    +private[spark] class OffHeapStore(blockManager: BlockManager, executorId: String)
    +  extends BlockStore(blockManager: BlockManager) with Logging {
    +
    +  lazy val offHeapManager: Option[OffHeapBlockManager] =
    +    OffHeapBlockManager.create(blockManager, executorId)
    +
    +  logInfo("OffHeap started")
    +
    +  override def getSize(blockId: BlockId): Long = {
    +    try {
    +      offHeapManager.map(_.getSize(blockId)).getOrElse(0)
    +    } catch {
    +      case NonFatal(t) => logError(s"error in getSize from $blockId")
    +        0
    +    }
    +  }
    +
    +  override def putBytes(blockId: BlockId, bytes: ByteBuffer, level: StorageLevel): PutResult = {
    +    putIntoOffHeapStore(blockId, bytes, returnValues = true)
    +  }
    +
    +  override def putArray(
    +      blockId: BlockId,
    +      values: Array[Any],
    +      level: StorageLevel,
    +      returnValues: Boolean): PutResult = {
    +    putIterator(blockId, values.toIterator, level, returnValues)
    +  }
    +
    +  override def putIterator(
    +      blockId: BlockId,
    +      values: Iterator[Any],
    +      level: StorageLevel,
    +      returnValues: Boolean): PutResult = {
    +    logDebug(s"Attempting to write values for block $blockId")
    +    val bytes = blockManager.dataSerialize(blockId, values)
    +    putIntoOffHeapStore(blockId, bytes, returnValues)
    +  }
    +
    +  private def putIntoOffHeapStore(
    +      blockId: BlockId,
    +      bytes: ByteBuffer,
    +      returnValues: Boolean): PutResult = {
    +
    +      // So that we do not modify the input offsets !
    +      // duplicate does not copy buffer, so inexpensive
    +      val byteBuffer = bytes.duplicate()
    +      byteBuffer.rewind()
    +      logDebug(s"Attempting to put block $blockId into OffHeap store")
    +      val startTime = System.currentTimeMillis
    +      // we should never hit here if offHeapManager is None. Handle it anyway for safety.
    +      try {
    +        if (offHeapManager.isDefined) {
    +          offHeapManager.get.putBytes(blockId, bytes)
    +          val finishTime = System.currentTimeMillis
    +          logDebug("Block %s stored as %s file in OffHeap store in %d ms".format(
    +            blockId, Utils.bytesToString(byteBuffer.limit), finishTime - startTime))
    +
    +          if (returnValues) {
    +            PutResult(bytes.limit(), Right(bytes.duplicate()))
    +          } else {
    +            PutResult(bytes.limit(), null)
    +          }
    +        } else {
    +          logError(s"error in putBytes $blockId")
    +          PutResult(bytes.limit(), null, Seq((blockId, BlockStatus.empty)))
    +        }
    +      } catch {
    +        case NonFatal(t) => logError(s"error in putBytes $blockId")
    +          PutResult(bytes.limit(), null, Seq((blockId, BlockStatus.empty)))
    +      }
    +  }
    +
    +  // We assume the block is removed even if exception thrown
    +  override def remove(blockId: BlockId): Boolean = {
    +    try {
    +      offHeapManager.map(_.removeFile(blockId)).getOrElse(true)
    +    } catch {
    +      case NonFatal(t) => logError(s"error in removing $blockId")
    +        true
    +    }
    +  }
    +
    +  override def getValues(blockId: BlockId): Option[Iterator[Any]] = {
    +    getBytes(blockId).map(buffer => blockManager.dataDeserialize(blockId, buffer))
    +  }
    +
    --- End diff --
    
    remove the extra blank line here when you update


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org