You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ma...@apache.org on 2013/09/01 23:58:58 UTC
[14/69] [abbrv] [partial] Initial work to rename package to
org.apache.spark
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/storage/BlockMessageArray.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/storage/BlockMessageArray.scala b/core/src/main/scala/spark/storage/BlockMessageArray.scala
deleted file mode 100644
index ee2fc16..0000000
--- a/core/src/main/scala/spark/storage/BlockMessageArray.scala
+++ /dev/null
@@ -1,159 +0,0 @@
-/*
- * 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 spark.storage
-
-import java.nio.ByteBuffer
-
-import scala.collection.mutable.ArrayBuffer
-
-import spark._
-import spark.network._
-
-private[spark]
-class BlockMessageArray(var blockMessages: Seq[BlockMessage]) extends Seq[BlockMessage] with Logging {
-
- def this(bm: BlockMessage) = this(Array(bm))
-
- def this() = this(null.asInstanceOf[Seq[BlockMessage]])
-
- def apply(i: Int) = blockMessages(i)
-
- def iterator = blockMessages.iterator
-
- def length = blockMessages.length
-
- initLogging()
-
- def set(bufferMessage: BufferMessage) {
- val startTime = System.currentTimeMillis
- val newBlockMessages = new ArrayBuffer[BlockMessage]()
- val buffer = bufferMessage.buffers(0)
- buffer.clear()
- /*
- println()
- println("BlockMessageArray: ")
- while(buffer.remaining > 0) {
- print(buffer.get())
- }
- buffer.rewind()
- println()
- println()
- */
- while (buffer.remaining() > 0) {
- val size = buffer.getInt()
- logDebug("Creating block message of size " + size + " bytes")
- val newBuffer = buffer.slice()
- newBuffer.clear()
- newBuffer.limit(size)
- logDebug("Trying to convert buffer " + newBuffer + " to block message")
- val newBlockMessage = BlockMessage.fromByteBuffer(newBuffer)
- logDebug("Created " + newBlockMessage)
- newBlockMessages += newBlockMessage
- buffer.position(buffer.position() + size)
- }
- val finishTime = System.currentTimeMillis
- logDebug("Converted block message array from buffer message in " + (finishTime - startTime) / 1000.0 + " s")
- this.blockMessages = newBlockMessages
- }
-
- def toBufferMessage: BufferMessage = {
- val buffers = new ArrayBuffer[ByteBuffer]()
-
- blockMessages.foreach(blockMessage => {
- val bufferMessage = blockMessage.toBufferMessage
- logDebug("Adding " + blockMessage)
- val sizeBuffer = ByteBuffer.allocate(4).putInt(bufferMessage.size)
- sizeBuffer.flip
- buffers += sizeBuffer
- buffers ++= bufferMessage.buffers
- logDebug("Added " + bufferMessage)
- })
-
- logDebug("Buffer list:")
- buffers.foreach((x: ByteBuffer) => logDebug("" + x))
- /*
- println()
- println("BlockMessageArray: ")
- buffers.foreach(b => {
- while(b.remaining > 0) {
- print(b.get())
- }
- b.rewind()
- })
- println()
- println()
- */
- return Message.createBufferMessage(buffers)
- }
-}
-
-private[spark] object BlockMessageArray {
-
- def fromBufferMessage(bufferMessage: BufferMessage): BlockMessageArray = {
- val newBlockMessageArray = new BlockMessageArray()
- newBlockMessageArray.set(bufferMessage)
- newBlockMessageArray
- }
-
- def main(args: Array[String]) {
- val blockMessages =
- (0 until 10).map { i =>
- if (i % 2 == 0) {
- val buffer = ByteBuffer.allocate(100)
- buffer.clear
- BlockMessage.fromPutBlock(PutBlock(i.toString, buffer, StorageLevel.MEMORY_ONLY_SER))
- } else {
- BlockMessage.fromGetBlock(GetBlock(i.toString))
- }
- }
- val blockMessageArray = new BlockMessageArray(blockMessages)
- println("Block message array created")
-
- val bufferMessage = blockMessageArray.toBufferMessage
- println("Converted to buffer message")
-
- val totalSize = bufferMessage.size
- val newBuffer = ByteBuffer.allocate(totalSize)
- newBuffer.clear()
- bufferMessage.buffers.foreach(buffer => {
- assert (0 == buffer.position())
- newBuffer.put(buffer)
- buffer.rewind()
- })
- newBuffer.flip
- val newBufferMessage = Message.createBufferMessage(newBuffer)
- println("Copied to new buffer message, size = " + newBufferMessage.size)
-
- val newBlockMessageArray = BlockMessageArray.fromBufferMessage(newBufferMessage)
- println("Converted back to block message array")
- newBlockMessageArray.foreach(blockMessage => {
- blockMessage.getType match {
- case BlockMessage.TYPE_PUT_BLOCK => {
- val pB = PutBlock(blockMessage.getId, blockMessage.getData, blockMessage.getLevel)
- println(pB)
- }
- case BlockMessage.TYPE_GET_BLOCK => {
- val gB = new GetBlock(blockMessage.getId)
- println(gB)
- }
- }
- })
- }
-}
-
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/storage/BlockObjectWriter.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/storage/BlockObjectWriter.scala b/core/src/main/scala/spark/storage/BlockObjectWriter.scala
deleted file mode 100644
index 3812009..0000000
--- a/core/src/main/scala/spark/storage/BlockObjectWriter.scala
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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 spark.storage
-
-
-/**
- * An interface for writing JVM objects to some underlying storage. This interface allows
- * appending data to an existing block, and can guarantee atomicity in the case of faults
- * as it allows the caller to revert partial writes.
- *
- * This interface does not support concurrent writes.
- */
-abstract class BlockObjectWriter(val blockId: String) {
-
- var closeEventHandler: () => Unit = _
-
- def open(): BlockObjectWriter
-
- def close() {
- closeEventHandler()
- }
-
- def isOpen: Boolean
-
- def registerCloseEventHandler(handler: () => Unit) {
- closeEventHandler = handler
- }
-
- /**
- * Flush the partial writes and commit them as a single atomic block. Return the
- * number of bytes written for this commit.
- */
- def commit(): Long
-
- /**
- * Reverts writes that haven't been flushed yet. Callers should invoke this function
- * when there are runtime exceptions.
- */
- def revertPartialWrites()
-
- /**
- * Writes an object.
- */
- def write(value: Any)
-
- /**
- * Size of the valid writes, in bytes.
- */
- def size(): Long
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/storage/BlockStore.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/storage/BlockStore.scala b/core/src/main/scala/spark/storage/BlockStore.scala
deleted file mode 100644
index c8db002..0000000
--- a/core/src/main/scala/spark/storage/BlockStore.scala
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * 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 spark.storage
-
-import java.nio.ByteBuffer
-import scala.collection.mutable.ArrayBuffer
-
-import spark.Logging
-
-/**
- * Abstract class to store blocks
- */
-private[spark]
-abstract class BlockStore(val blockManager: BlockManager) extends Logging {
- def putBytes(blockId: String, bytes: ByteBuffer, level: StorageLevel)
-
- /**
- * Put in a block and, possibly, also return its content as either bytes or another Iterator.
- * This is used to efficiently write the values to multiple locations (e.g. for replication).
- *
- * @return a PutResult that contains the size of the data, as well as the values put if
- * returnValues is true (if not, the result's data field can be null)
- */
- def putValues(blockId: String, values: ArrayBuffer[Any], level: StorageLevel,
- returnValues: Boolean) : PutResult
-
- /**
- * Return the size of a block in bytes.
- */
- def getSize(blockId: String): Long
-
- def getBytes(blockId: String): Option[ByteBuffer]
-
- def getValues(blockId: String): Option[Iterator[Any]]
-
- /**
- * Remove a block, if it exists.
- * @param blockId the block to remove.
- * @return True if the block was found and removed, False otherwise.
- */
- def remove(blockId: String): Boolean
-
- def contains(blockId: String): Boolean
-
- def clear() { }
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/storage/DiskStore.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/storage/DiskStore.scala b/core/src/main/scala/spark/storage/DiskStore.scala
deleted file mode 100644
index b144971..0000000
--- a/core/src/main/scala/spark/storage/DiskStore.scala
+++ /dev/null
@@ -1,329 +0,0 @@
-/*
- * 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 spark.storage
-
-import java.io.{File, FileOutputStream, OutputStream, RandomAccessFile}
-import java.nio.ByteBuffer
-import java.nio.channels.FileChannel
-import java.nio.channels.FileChannel.MapMode
-import java.util.{Random, Date}
-import java.text.SimpleDateFormat
-
-import scala.collection.mutable.ArrayBuffer
-
-import it.unimi.dsi.fastutil.io.FastBufferedOutputStream
-
-import spark.Utils
-import spark.executor.ExecutorExitCode
-import spark.serializer.{Serializer, SerializationStream}
-import spark.Logging
-import spark.network.netty.ShuffleSender
-import spark.network.netty.PathResolver
-
-
-/**
- * Stores BlockManager blocks on disk.
- */
-private class DiskStore(blockManager: BlockManager, rootDirs: String)
- extends BlockStore(blockManager) with Logging {
-
- class DiskBlockObjectWriter(blockId: String, serializer: Serializer, bufferSize: Int)
- extends BlockObjectWriter(blockId) {
-
- private val f: File = createFile(blockId /*, allowAppendExisting */)
-
- // The file channel, used for repositioning / truncating the file.
- private var channel: FileChannel = null
- private var bs: OutputStream = null
- private var objOut: SerializationStream = null
- private var lastValidPosition = 0L
- private var initialized = false
-
- override def open(): DiskBlockObjectWriter = {
- val fos = new FileOutputStream(f, true)
- channel = fos.getChannel()
- bs = blockManager.wrapForCompression(blockId, new FastBufferedOutputStream(fos, bufferSize))
- objOut = serializer.newInstance().serializeStream(bs)
- initialized = true
- this
- }
-
- override def close() {
- if (initialized) {
- objOut.close()
- channel = null
- bs = null
- objOut = null
- }
- // Invoke the close callback handler.
- super.close()
- }
-
- override def isOpen: Boolean = objOut != null
-
- // Flush the partial writes, and set valid length to be the length of the entire file.
- // Return the number of bytes written for this commit.
- override def commit(): Long = {
- if (initialized) {
- // NOTE: Flush the serializer first and then the compressed/buffered output stream
- objOut.flush()
- bs.flush()
- val prevPos = lastValidPosition
- lastValidPosition = channel.position()
- lastValidPosition - prevPos
- } else {
- // lastValidPosition is zero if stream is uninitialized
- lastValidPosition
- }
- }
-
- override def revertPartialWrites() {
- if (initialized) {
- // Discard current writes. We do this by flushing the outstanding writes and
- // truncate the file to the last valid position.
- objOut.flush()
- bs.flush()
- channel.truncate(lastValidPosition)
- }
- }
-
- override def write(value: Any) {
- if (!initialized) {
- open()
- }
- objOut.writeObject(value)
- }
-
- override def size(): Long = lastValidPosition
- }
-
- private val MAX_DIR_CREATION_ATTEMPTS: Int = 10
- private val subDirsPerLocalDir = System.getProperty("spark.diskStore.subDirectories", "64").toInt
-
- private var shuffleSender : ShuffleSender = null
- // Create one local directory for each path mentioned in spark.local.dir; then, inside this
- // directory, create multiple subdirectories that we will hash files into, in order to avoid
- // having really large inodes at the top level.
- private val localDirs: Array[File] = createLocalDirs()
- private val subDirs = Array.fill(localDirs.length)(new Array[File](subDirsPerLocalDir))
-
- addShutdownHook()
-
- def getBlockWriter(blockId: String, serializer: Serializer, bufferSize: Int)
- : BlockObjectWriter = {
- new DiskBlockObjectWriter(blockId, serializer, bufferSize)
- }
-
- override def getSize(blockId: String): Long = {
- getFile(blockId).length()
- }
-
- override def putBytes(blockId: String, _bytes: ByteBuffer, level: StorageLevel) {
- // So that we do not modify the input offsets !
- // duplicate does not copy buffer, so inexpensive
- val bytes = _bytes.duplicate()
- logDebug("Attempting to put block " + blockId)
- val startTime = System.currentTimeMillis
- val file = createFile(blockId)
- val channel = new RandomAccessFile(file, "rw").getChannel()
- while (bytes.remaining > 0) {
- channel.write(bytes)
- }
- channel.close()
- val finishTime = System.currentTimeMillis
- logDebug("Block %s stored as %s file on disk in %d ms".format(
- blockId, Utils.bytesToString(bytes.limit), (finishTime - startTime)))
- }
-
- private def getFileBytes(file: File): ByteBuffer = {
- val length = file.length()
- val channel = new RandomAccessFile(file, "r").getChannel()
- val buffer = try {
- channel.map(MapMode.READ_ONLY, 0, length)
- } finally {
- channel.close()
- }
-
- buffer
- }
-
- override def putValues(
- blockId: String,
- values: ArrayBuffer[Any],
- level: StorageLevel,
- returnValues: Boolean)
- : PutResult = {
-
- logDebug("Attempting to write values for block " + blockId)
- val startTime = System.currentTimeMillis
- val file = createFile(blockId)
- val fileOut = blockManager.wrapForCompression(blockId,
- new FastBufferedOutputStream(new FileOutputStream(file)))
- val objOut = blockManager.defaultSerializer.newInstance().serializeStream(fileOut)
- objOut.writeAll(values.iterator)
- objOut.close()
- val length = file.length()
-
- val timeTaken = System.currentTimeMillis - startTime
- logDebug("Block %s stored as %s file on disk in %d ms".format(
- blockId, Utils.bytesToString(length), timeTaken))
-
- if (returnValues) {
- // Return a byte buffer for the contents of the file
- val buffer = getFileBytes(file)
- PutResult(length, Right(buffer))
- } else {
- PutResult(length, null)
- }
- }
-
- override def getBytes(blockId: String): Option[ByteBuffer] = {
- val file = getFile(blockId)
- val bytes = getFileBytes(file)
- Some(bytes)
- }
-
- override def getValues(blockId: String): Option[Iterator[Any]] = {
- getBytes(blockId).map(bytes => blockManager.dataDeserialize(blockId, bytes))
- }
-
- /**
- * A version of getValues that allows a custom serializer. This is used as part of the
- * shuffle short-circuit code.
- */
- def getValues(blockId: String, serializer: Serializer): Option[Iterator[Any]] = {
- getBytes(blockId).map(bytes => blockManager.dataDeserialize(blockId, bytes, serializer))
- }
-
- override def remove(blockId: String): Boolean = {
- val file = getFile(blockId)
- if (file.exists()) {
- file.delete()
- } else {
- false
- }
- }
-
- override def contains(blockId: String): Boolean = {
- getFile(blockId).exists()
- }
-
- private def createFile(blockId: String, allowAppendExisting: Boolean = false): File = {
- val file = getFile(blockId)
- if (!allowAppendExisting && file.exists()) {
- // NOTE(shivaram): Delete the file if it exists. This might happen if a ShuffleMap task
- // was rescheduled on the same machine as the old task.
- logWarning("File for block " + blockId + " already exists on disk: " + file + ". Deleting")
- file.delete()
- }
- file
- }
-
- private def getFile(blockId: String): File = {
- logDebug("Getting file for block " + blockId)
-
- // Figure out which local directory it hashes to, and which subdirectory in that
- val hash = math.abs(blockId.hashCode)
- val dirId = hash % localDirs.length
- val subDirId = (hash / localDirs.length) % subDirsPerLocalDir
-
- // Create the subdirectory if it doesn't already exist
- var subDir = subDirs(dirId)(subDirId)
- if (subDir == null) {
- subDir = subDirs(dirId).synchronized {
- val old = subDirs(dirId)(subDirId)
- if (old != null) {
- old
- } else {
- val newDir = new File(localDirs(dirId), "%02x".format(subDirId))
- newDir.mkdir()
- subDirs(dirId)(subDirId) = newDir
- newDir
- }
- }
- }
-
- new File(subDir, blockId)
- }
-
- private def createLocalDirs(): Array[File] = {
- logDebug("Creating local directories at root dirs '" + rootDirs + "'")
- val dateFormat = new SimpleDateFormat("yyyyMMddHHmmss")
- rootDirs.split(",").map { rootDir =>
- var foundLocalDir = false
- var localDir: File = null
- var localDirId: String = null
- var tries = 0
- val rand = new Random()
- while (!foundLocalDir && tries < MAX_DIR_CREATION_ATTEMPTS) {
- tries += 1
- try {
- localDirId = "%s-%04x".format(dateFormat.format(new Date), rand.nextInt(65536))
- localDir = new File(rootDir, "spark-local-" + localDirId)
- if (!localDir.exists) {
- foundLocalDir = localDir.mkdirs()
- }
- } catch {
- case e: Exception =>
- logWarning("Attempt " + tries + " to create local dir " + localDir + " failed", e)
- }
- }
- if (!foundLocalDir) {
- logError("Failed " + MAX_DIR_CREATION_ATTEMPTS +
- " attempts to create local dir in " + rootDir)
- System.exit(ExecutorExitCode.DISK_STORE_FAILED_TO_CREATE_DIR)
- }
- logInfo("Created local directory at " + localDir)
- localDir
- }
- }
-
- private def addShutdownHook() {
- localDirs.foreach(localDir => Utils.registerShutdownDeleteDir(localDir))
- Runtime.getRuntime.addShutdownHook(new Thread("delete Spark local dirs") {
- override def run() {
- logDebug("Shutdown hook called")
- localDirs.foreach { localDir =>
- try {
- if (!Utils.hasRootAsShutdownDeleteDir(localDir)) Utils.deleteRecursively(localDir)
- } catch {
- case t: Throwable =>
- logError("Exception while deleting local spark dir: " + localDir, t)
- }
- }
- if (shuffleSender != null) {
- shuffleSender.stop
- }
- }
- })
- }
-
- private[storage] def startShuffleBlockSender(port: Int): Int = {
- val pResolver = new PathResolver {
- override def getAbsolutePath(blockId: String): String = {
- if (!blockId.startsWith("shuffle_")) {
- return null
- }
- DiskStore.this.getFile(blockId).getAbsolutePath()
- }
- }
- shuffleSender = new ShuffleSender(port, pResolver)
- logInfo("Created ShuffleSender binding to port : "+ shuffleSender.port)
- shuffleSender.port
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/storage/MemoryStore.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/storage/MemoryStore.scala b/core/src/main/scala/spark/storage/MemoryStore.scala
deleted file mode 100644
index 5a51f5c..0000000
--- a/core/src/main/scala/spark/storage/MemoryStore.scala
+++ /dev/null
@@ -1,257 +0,0 @@
-/*
- * 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 spark.storage
-
-import java.util.LinkedHashMap
-import java.util.concurrent.ArrayBlockingQueue
-import spark.{SizeEstimator, Utils}
-import java.nio.ByteBuffer
-import collection.mutable.ArrayBuffer
-
-/**
- * Stores blocks in memory, either as ArrayBuffers of deserialized Java objects or as
- * serialized ByteBuffers.
- */
-private class MemoryStore(blockManager: BlockManager, maxMemory: Long)
- extends BlockStore(blockManager) {
-
- case class Entry(value: Any, size: Long, deserialized: Boolean, var dropPending: Boolean = false)
-
- private val entries = new LinkedHashMap[String, Entry](32, 0.75f, true)
- private var currentMemory = 0L
- // Object used to ensure that only one thread is putting blocks and if necessary, dropping
- // blocks from the memory store.
- private val putLock = new Object()
-
- logInfo("MemoryStore started with capacity %s.".format(Utils.bytesToString(maxMemory)))
-
- def freeMemory: Long = maxMemory - currentMemory
-
- override def getSize(blockId: String): Long = {
- entries.synchronized {
- entries.get(blockId).size
- }
- }
-
- override def putBytes(blockId: String, _bytes: ByteBuffer, level: StorageLevel) {
- // Work on a duplicate - since the original input might be used elsewhere.
- val bytes = _bytes.duplicate()
- bytes.rewind()
- if (level.deserialized) {
- val values = blockManager.dataDeserialize(blockId, bytes)
- val elements = new ArrayBuffer[Any]
- elements ++= values
- val sizeEstimate = SizeEstimator.estimate(elements.asInstanceOf[AnyRef])
- tryToPut(blockId, elements, sizeEstimate, true)
- } else {
- tryToPut(blockId, bytes, bytes.limit, false)
- }
- }
-
- override def putValues(
- blockId: String,
- values: ArrayBuffer[Any],
- level: StorageLevel,
- returnValues: Boolean)
- : PutResult = {
-
- if (level.deserialized) {
- val sizeEstimate = SizeEstimator.estimate(values.asInstanceOf[AnyRef])
- tryToPut(blockId, values, sizeEstimate, true)
- PutResult(sizeEstimate, Left(values.iterator))
- } else {
- val bytes = blockManager.dataSerialize(blockId, values.iterator)
- tryToPut(blockId, bytes, bytes.limit, false)
- PutResult(bytes.limit(), Right(bytes.duplicate()))
- }
- }
-
- override def getBytes(blockId: String): Option[ByteBuffer] = {
- val entry = entries.synchronized {
- entries.get(blockId)
- }
- if (entry == null) {
- None
- } else if (entry.deserialized) {
- Some(blockManager.dataSerialize(blockId, entry.value.asInstanceOf[ArrayBuffer[Any]].iterator))
- } else {
- Some(entry.value.asInstanceOf[ByteBuffer].duplicate()) // Doesn't actually copy the data
- }
- }
-
- override def getValues(blockId: String): Option[Iterator[Any]] = {
- val entry = entries.synchronized {
- entries.get(blockId)
- }
- if (entry == null) {
- None
- } else if (entry.deserialized) {
- Some(entry.value.asInstanceOf[ArrayBuffer[Any]].iterator)
- } else {
- val buffer = entry.value.asInstanceOf[ByteBuffer].duplicate() // Doesn't actually copy data
- Some(blockManager.dataDeserialize(blockId, buffer))
- }
- }
-
- override def remove(blockId: String): Boolean = {
- entries.synchronized {
- val entry = entries.get(blockId)
- if (entry != null) {
- entries.remove(blockId)
- currentMemory -= entry.size
- logInfo("Block %s of size %d dropped from memory (free %d)".format(
- blockId, entry.size, freeMemory))
- true
- } else {
- false
- }
- }
- }
-
- override def clear() {
- entries.synchronized {
- entries.clear()
- }
- logInfo("MemoryStore cleared")
- }
-
- /**
- * Return the RDD ID that a given block ID is from, or null if it is not an RDD block.
- */
- private def getRddId(blockId: String): String = {
- if (blockId.startsWith("rdd_")) {
- blockId.split('_')(1)
- } else {
- null
- }
- }
-
- /**
- * Try to put in a set of values, if we can free up enough space. The value should either be
- * an ArrayBuffer if deserialized is true or a ByteBuffer otherwise. Its (possibly estimated)
- * size must also be passed by the caller.
- *
- * Locks on the object putLock to ensure that all the put requests and its associated block
- * dropping is done by only on thread at a time. Otherwise while one thread is dropping
- * blocks to free memory for one block, another thread may use up the freed space for
- * another block.
- */
- private def tryToPut(blockId: String, value: Any, size: Long, deserialized: Boolean): Boolean = {
- // TODO: Its possible to optimize the locking by locking entries only when selecting blocks
- // to be dropped. Once the to-be-dropped blocks have been selected, and lock on entries has been
- // released, it must be ensured that those to-be-dropped blocks are not double counted for
- // freeing up more space for another block that needs to be put. Only then the actually dropping
- // of blocks (and writing to disk if necessary) can proceed in parallel.
- putLock.synchronized {
- if (ensureFreeSpace(blockId, size)) {
- val entry = new Entry(value, size, deserialized)
- entries.synchronized { entries.put(blockId, entry) }
- currentMemory += size
- if (deserialized) {
- logInfo("Block %s stored as values to memory (estimated size %s, free %s)".format(
- blockId, Utils.bytesToString(size), Utils.bytesToString(freeMemory)))
- } else {
- logInfo("Block %s stored as bytes to memory (size %s, free %s)".format(
- blockId, Utils.bytesToString(size), Utils.bytesToString(freeMemory)))
- }
- true
- } else {
- // Tell the block manager that we couldn't put it in memory so that it can drop it to
- // disk if the block allows disk storage.
- val data = if (deserialized) {
- Left(value.asInstanceOf[ArrayBuffer[Any]])
- } else {
- Right(value.asInstanceOf[ByteBuffer].duplicate())
- }
- blockManager.dropFromMemory(blockId, data)
- false
- }
- }
- }
-
- /**
- * Tries to free up a given amount of space to store a particular block, but can fail and return
- * false if either the block is bigger than our memory or it would require replacing another
- * block from the same RDD (which leads to a wasteful cyclic replacement pattern for RDDs that
- * don't fit into memory that we want to avoid).
- *
- * Assumes that a lock is held by the caller to ensure only one thread is dropping blocks.
- * Otherwise, the freed space may fill up before the caller puts in their new value.
- */
- private def ensureFreeSpace(blockIdToAdd: String, space: Long): Boolean = {
-
- logInfo("ensureFreeSpace(%d) called with curMem=%d, maxMem=%d".format(
- space, currentMemory, maxMemory))
-
- if (space > maxMemory) {
- logInfo("Will not store " + blockIdToAdd + " as it is larger than our memory limit")
- return false
- }
-
- if (maxMemory - currentMemory < space) {
- val rddToAdd = getRddId(blockIdToAdd)
- val selectedBlocks = new ArrayBuffer[String]()
- var selectedMemory = 0L
-
- // This is synchronized to ensure that the set of entries is not changed
- // (because of getValue or getBytes) while traversing the iterator, as that
- // can lead to exceptions.
- entries.synchronized {
- val iterator = entries.entrySet().iterator()
- while (maxMemory - (currentMemory - selectedMemory) < space && iterator.hasNext) {
- val pair = iterator.next()
- val blockId = pair.getKey
- if (rddToAdd != null && rddToAdd == getRddId(blockId)) {
- logInfo("Will not store " + blockIdToAdd + " as it would require dropping another " +
- "block from the same RDD")
- return false
- }
- selectedBlocks += blockId
- selectedMemory += pair.getValue.size
- }
- }
-
- if (maxMemory - (currentMemory - selectedMemory) >= space) {
- logInfo(selectedBlocks.size + " blocks selected for dropping")
- for (blockId <- selectedBlocks) {
- val entry = entries.synchronized { entries.get(blockId) }
- // This should never be null as only one thread should be dropping
- // blocks and removing entries. However the check is still here for
- // future safety.
- if (entry != null) {
- val data = if (entry.deserialized) {
- Left(entry.value.asInstanceOf[ArrayBuffer[Any]])
- } else {
- Right(entry.value.asInstanceOf[ByteBuffer].duplicate())
- }
- blockManager.dropFromMemory(blockId, data)
- }
- }
- return true
- } else {
- return false
- }
- }
- return true
- }
-
- override def contains(blockId: String): Boolean = {
- entries.synchronized { entries.containsKey(blockId) }
- }
-}
-
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/storage/PutResult.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/storage/PutResult.scala b/core/src/main/scala/spark/storage/PutResult.scala
deleted file mode 100644
index 3a0974f..0000000
--- a/core/src/main/scala/spark/storage/PutResult.scala
+++ /dev/null
@@ -1,26 +0,0 @@
-/*
- * 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 spark.storage
-
-import java.nio.ByteBuffer
-
-/**
- * Result of adding a block into a BlockStore. Contains its estimated size, and possibly the
- * values put if the caller asked for them to be returned (e.g. for chaining replication)
- */
-private[spark] case class PutResult(size: Long, data: Either[Iterator[_], ByteBuffer])
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/storage/ShuffleBlockManager.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/storage/ShuffleBlockManager.scala b/core/src/main/scala/spark/storage/ShuffleBlockManager.scala
deleted file mode 100644
index 8a7a6f9..0000000
--- a/core/src/main/scala/spark/storage/ShuffleBlockManager.scala
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
- * 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 spark.storage
-
-import spark.serializer.Serializer
-
-
-private[spark]
-class ShuffleWriterGroup(val id: Int, val writers: Array[BlockObjectWriter])
-
-
-private[spark]
-trait ShuffleBlocks {
- def acquireWriters(mapId: Int): ShuffleWriterGroup
- def releaseWriters(group: ShuffleWriterGroup)
-}
-
-
-private[spark]
-class ShuffleBlockManager(blockManager: BlockManager) {
-
- def forShuffle(shuffleId: Int, numBuckets: Int, serializer: Serializer): ShuffleBlocks = {
- new ShuffleBlocks {
- // Get a group of writers for a map task.
- override def acquireWriters(mapId: Int): ShuffleWriterGroup = {
- val bufferSize = System.getProperty("spark.shuffle.file.buffer.kb", "100").toInt * 1024
- val writers = Array.tabulate[BlockObjectWriter](numBuckets) { bucketId =>
- val blockId = ShuffleBlockManager.blockId(shuffleId, bucketId, mapId)
- blockManager.getDiskBlockWriter(blockId, serializer, bufferSize)
- }
- new ShuffleWriterGroup(mapId, writers)
- }
-
- override def releaseWriters(group: ShuffleWriterGroup) = {
- // Nothing really to release here.
- }
- }
- }
-}
-
-
-private[spark]
-object ShuffleBlockManager {
-
- // Returns the block id for a given shuffle block.
- def blockId(shuffleId: Int, bucketId: Int, groupId: Int): String = {
- "shuffle_" + shuffleId + "_" + groupId + "_" + bucketId
- }
-
- // Returns true if the block is a shuffle block.
- def isShuffle(blockId: String): Boolean = blockId.startsWith("shuffle_")
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/storage/StorageLevel.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/storage/StorageLevel.scala b/core/src/main/scala/spark/storage/StorageLevel.scala
deleted file mode 100644
index f526509..0000000
--- a/core/src/main/scala/spark/storage/StorageLevel.scala
+++ /dev/null
@@ -1,146 +0,0 @@
-/*
- * 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 spark.storage
-
-import java.io.{Externalizable, IOException, ObjectInput, ObjectOutput}
-
-/**
- * Flags for controlling the storage of an RDD. Each StorageLevel records whether to use memory,
- * whether to drop the RDD to disk if it falls out of memory, whether to keep the data in memory
- * in a serialized format, and whether to replicate the RDD partitions on multiple nodes.
- * The [[spark.storage.StorageLevel$]] singleton object contains some static constants for
- * commonly useful storage levels. To create your own storage level object, use the factor method
- * of the singleton object (`StorageLevel(...)`).
- */
-class StorageLevel private(
- private var useDisk_ : Boolean,
- private var useMemory_ : Boolean,
- private var deserialized_ : Boolean,
- private var replication_ : Int = 1)
- extends Externalizable {
-
- // TODO: Also add fields for caching priority, dataset ID, and flushing.
- private def this(flags: Int, replication: Int) {
- this((flags & 4) != 0, (flags & 2) != 0, (flags & 1) != 0, replication)
- }
-
- def this() = this(false, true, false) // For deserialization
-
- def useDisk = useDisk_
- def useMemory = useMemory_
- def deserialized = deserialized_
- def replication = replication_
-
- assert(replication < 40, "Replication restricted to be less than 40 for calculating hashcodes")
-
- override def clone(): StorageLevel = new StorageLevel(
- this.useDisk, this.useMemory, this.deserialized, this.replication)
-
- override def equals(other: Any): Boolean = other match {
- case s: StorageLevel =>
- s.useDisk == useDisk &&
- s.useMemory == useMemory &&
- s.deserialized == deserialized &&
- s.replication == replication
- case _ =>
- false
- }
-
- def isValid = ((useMemory || useDisk) && (replication > 0))
-
- def toInt: Int = {
- var ret = 0
- if (useDisk_) {
- ret |= 4
- }
- if (useMemory_) {
- ret |= 2
- }
- if (deserialized_) {
- ret |= 1
- }
- return ret
- }
-
- override def writeExternal(out: ObjectOutput) {
- out.writeByte(toInt)
- out.writeByte(replication_)
- }
-
- override def readExternal(in: ObjectInput) {
- val flags = in.readByte()
- useDisk_ = (flags & 4) != 0
- useMemory_ = (flags & 2) != 0
- deserialized_ = (flags & 1) != 0
- replication_ = in.readByte()
- }
-
- @throws(classOf[IOException])
- private def readResolve(): Object = StorageLevel.getCachedStorageLevel(this)
-
- override def toString: String =
- "StorageLevel(%b, %b, %b, %d)".format(useDisk, useMemory, deserialized, replication)
-
- override def hashCode(): Int = toInt * 41 + replication
- def description : String = {
- var result = ""
- result += (if (useDisk) "Disk " else "")
- result += (if (useMemory) "Memory " else "")
- result += (if (deserialized) "Deserialized " else "Serialized")
- result += "%sx Replicated".format(replication)
- result
- }
-}
-
-
-object StorageLevel {
- val NONE = new StorageLevel(false, false, false)
- val DISK_ONLY = new StorageLevel(true, false, false)
- val DISK_ONLY_2 = new StorageLevel(true, false, false, 2)
- val MEMORY_ONLY = new StorageLevel(false, true, true)
- val MEMORY_ONLY_2 = new StorageLevel(false, true, true, 2)
- val MEMORY_ONLY_SER = new StorageLevel(false, true, false)
- val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, 2)
- val MEMORY_AND_DISK = new StorageLevel(true, true, true)
- val MEMORY_AND_DISK_2 = new StorageLevel(true, true, true, 2)
- val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false)
- val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, 2)
-
- /** Create a new StorageLevel object */
- def apply(useDisk: Boolean, useMemory: Boolean, deserialized: Boolean, replication: Int = 1) =
- getCachedStorageLevel(new StorageLevel(useDisk, useMemory, deserialized, replication))
-
- /** Create a new StorageLevel object from its integer representation */
- def apply(flags: Int, replication: Int) =
- getCachedStorageLevel(new StorageLevel(flags, replication))
-
- /** Read StorageLevel object from ObjectInput stream */
- def apply(in: ObjectInput) = {
- val obj = new StorageLevel()
- obj.readExternal(in)
- getCachedStorageLevel(obj)
- }
-
- private[spark]
- val storageLevelCache = new java.util.concurrent.ConcurrentHashMap[StorageLevel, StorageLevel]()
-
- private[spark] def getCachedStorageLevel(level: StorageLevel): StorageLevel = {
- storageLevelCache.putIfAbsent(level, level)
- storageLevelCache.get(level)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/storage/StorageUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/storage/StorageUtils.scala b/core/src/main/scala/spark/storage/StorageUtils.scala
deleted file mode 100644
index 123b8f6..0000000
--- a/core/src/main/scala/spark/storage/StorageUtils.scala
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * 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 spark.storage
-
-import spark.{Utils, SparkContext}
-import BlockManagerMasterActor.BlockStatus
-
-private[spark]
-case class StorageStatus(blockManagerId: BlockManagerId, maxMem: Long,
- blocks: Map[String, BlockStatus]) {
-
- def memUsed(blockPrefix: String = "") = {
- blocks.filterKeys(_.startsWith(blockPrefix)).values.map(_.memSize).
- reduceOption(_+_).getOrElse(0l)
- }
-
- def diskUsed(blockPrefix: String = "") = {
- blocks.filterKeys(_.startsWith(blockPrefix)).values.map(_.diskSize).
- reduceOption(_+_).getOrElse(0l)
- }
-
- def memRemaining : Long = maxMem - memUsed()
-
-}
-
-case class RDDInfo(id: Int, name: String, storageLevel: StorageLevel,
- numCachedPartitions: Int, numPartitions: Int, memSize: Long, diskSize: Long)
- extends Ordered[RDDInfo] {
- override def toString = {
- import Utils.bytesToString
- "RDD \"%s\" (%d) Storage: %s; CachedPartitions: %d; TotalPartitions: %d; MemorySize: %s; DiskSize: %s".format(name, id,
- storageLevel.toString, numCachedPartitions, numPartitions, bytesToString(memSize), bytesToString(diskSize))
- }
-
- override def compare(that: RDDInfo) = {
- this.id - that.id
- }
-}
-
-/* Helper methods for storage-related objects */
-private[spark]
-object StorageUtils {
-
- /* Returns RDD-level information, compiled from a list of StorageStatus objects */
- def rddInfoFromStorageStatus(storageStatusList: Seq[StorageStatus],
- sc: SparkContext) : Array[RDDInfo] = {
- rddInfoFromBlockStatusList(storageStatusList.flatMap(_.blocks).toMap, sc)
- }
-
- /* Returns a map of blocks to their locations, compiled from a list of StorageStatus objects */
- def blockLocationsFromStorageStatus(storageStatusList: Seq[StorageStatus]) = {
- val blockLocationPairs = storageStatusList
- .flatMap(s => s.blocks.map(b => (b._1, s.blockManagerId.hostPort)))
- blockLocationPairs.groupBy(_._1).map{case (k, v) => (k, v.unzip._2)}.toMap
- }
-
- /* Given a list of BlockStatus objets, returns information for each RDD */
- def rddInfoFromBlockStatusList(infos: Map[String, BlockStatus],
- sc: SparkContext) : Array[RDDInfo] = {
-
- // Group by rddId, ignore the partition name
- val groupedRddBlocks = infos.filterKeys(_.startsWith("rdd_")).groupBy { case(k, v) =>
- k.substring(0,k.lastIndexOf('_'))
- }.mapValues(_.values.toArray)
-
- // For each RDD, generate an RDDInfo object
- val rddInfos = groupedRddBlocks.map { case (rddKey, rddBlocks) =>
- // Add up memory and disk sizes
- val memSize = rddBlocks.map(_.memSize).reduce(_ + _)
- val diskSize = rddBlocks.map(_.diskSize).reduce(_ + _)
-
- // Find the id of the RDD, e.g. rdd_1 => 1
- val rddId = rddKey.split("_").last.toInt
-
- // Get the friendly name and storage level for the RDD, if available
- sc.persistentRdds.get(rddId).map { r =>
- val rddName = Option(r.name).getOrElse(rddKey)
- val rddStorageLevel = r.getStorageLevel
- RDDInfo(rddId, rddName, rddStorageLevel, rddBlocks.length, r.partitions.size, memSize, diskSize)
- }
- }.flatten.toArray
-
- scala.util.Sorting.quickSort(rddInfos)
-
- rddInfos
- }
-
- /* Removes all BlockStatus object that are not part of a block prefix */
- def filterStorageStatusByPrefix(storageStatusList: Array[StorageStatus],
- prefix: String) : Array[StorageStatus] = {
-
- storageStatusList.map { status =>
- val newBlocks = status.blocks.filterKeys(_.startsWith(prefix))
- //val newRemainingMem = status.maxMem - newBlocks.values.map(_.memSize).reduce(_ + _)
- StorageStatus(status.blockManagerId, status.maxMem, newBlocks)
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/storage/ThreadingTest.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/storage/ThreadingTest.scala b/core/src/main/scala/spark/storage/ThreadingTest.scala
deleted file mode 100644
index b3ab1ff..0000000
--- a/core/src/main/scala/spark/storage/ThreadingTest.scala
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * 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 spark.storage
-
-import akka.actor._
-
-import spark.KryoSerializer
-import java.util.concurrent.ArrayBlockingQueue
-import util.Random
-
-/**
- * This class tests the BlockManager and MemoryStore for thread safety and
- * deadlocks. It spawns a number of producer and consumer threads. Producer
- * threads continuously pushes blocks into the BlockManager and consumer
- * threads continuously retrieves the blocks form the BlockManager and tests
- * whether the block is correct or not.
- */
-private[spark] object ThreadingTest {
-
- val numProducers = 5
- val numBlocksPerProducer = 20000
-
- private[spark] class ProducerThread(manager: BlockManager, id: Int) extends Thread {
- val queue = new ArrayBlockingQueue[(String, Seq[Int])](100)
-
- override def run() {
- for (i <- 1 to numBlocksPerProducer) {
- val blockId = "b-" + id + "-" + i
- val blockSize = Random.nextInt(1000)
- val block = (1 to blockSize).map(_ => Random.nextInt())
- val level = randomLevel()
- val startTime = System.currentTimeMillis()
- manager.put(blockId, block.iterator, level, true)
- println("Pushed block " + blockId + " in " + (System.currentTimeMillis - startTime) + " ms")
- queue.add((blockId, block))
- }
- println("Producer thread " + id + " terminated")
- }
-
- def randomLevel(): StorageLevel = {
- math.abs(Random.nextInt()) % 4 match {
- case 0 => StorageLevel.MEMORY_ONLY
- case 1 => StorageLevel.MEMORY_ONLY_SER
- case 2 => StorageLevel.MEMORY_AND_DISK
- case 3 => StorageLevel.MEMORY_AND_DISK_SER
- }
- }
- }
-
- private[spark] class ConsumerThread(
- manager: BlockManager,
- queue: ArrayBlockingQueue[(String, Seq[Int])]
- ) extends Thread {
- var numBlockConsumed = 0
-
- override def run() {
- println("Consumer thread started")
- while(numBlockConsumed < numBlocksPerProducer) {
- val (blockId, block) = queue.take()
- val startTime = System.currentTimeMillis()
- manager.get(blockId) match {
- case Some(retrievedBlock) =>
- assert(retrievedBlock.toList.asInstanceOf[List[Int]] == block.toList,
- "Block " + blockId + " did not match")
- println("Got block " + blockId + " in " +
- (System.currentTimeMillis - startTime) + " ms")
- case None =>
- assert(false, "Block " + blockId + " could not be retrieved")
- }
- numBlockConsumed += 1
- }
- println("Consumer thread terminated")
- }
- }
-
- def main(args: Array[String]) {
- System.setProperty("spark.kryoserializer.buffer.mb", "1")
- val actorSystem = ActorSystem("test")
- val serializer = new KryoSerializer
- val blockManagerMaster = new BlockManagerMaster(
- actorSystem.actorOf(Props(new BlockManagerMasterActor(true))))
- val blockManager = new BlockManager(
- "<driver>", actorSystem, blockManagerMaster, serializer, 1024 * 1024)
- val producers = (1 to numProducers).map(i => new ProducerThread(blockManager, i))
- val consumers = producers.map(p => new ConsumerThread(blockManager, p.queue))
- producers.foreach(_.start)
- consumers.foreach(_.start)
- producers.foreach(_.join)
- consumers.foreach(_.join)
- blockManager.stop()
- blockManagerMaster.stop()
- actorSystem.shutdown()
- actorSystem.awaitTermination()
- println("Everything stopped.")
- println(
- "It will take sometime for the JVM to clean all temporary files and shutdown. Sit tight.")
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/ui/JettyUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/ui/JettyUtils.scala b/core/src/main/scala/spark/ui/JettyUtils.scala
deleted file mode 100644
index f66fe39..0000000
--- a/core/src/main/scala/spark/ui/JettyUtils.scala
+++ /dev/null
@@ -1,132 +0,0 @@
-/*
- * 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 spark.ui
-
-import javax.servlet.http.{HttpServletResponse, HttpServletRequest}
-
-import scala.annotation.tailrec
-import scala.util.{Try, Success, Failure}
-import scala.xml.Node
-
-import net.liftweb.json.{JValue, pretty, render}
-
-import org.eclipse.jetty.server.{Server, Request, Handler}
-import org.eclipse.jetty.server.handler.{ResourceHandler, HandlerList, ContextHandler, AbstractHandler}
-import org.eclipse.jetty.util.thread.QueuedThreadPool
-
-import spark.Logging
-
-
-/** Utilities for launching a web server using Jetty's HTTP Server class */
-private[spark] object JettyUtils extends Logging {
- // Base type for a function that returns something based on an HTTP request. Allows for
- // implicit conversion from many types of functions to jetty Handlers.
- type Responder[T] = HttpServletRequest => T
-
- // Conversions from various types of Responder's to jetty Handlers
- implicit def jsonResponderToHandler(responder: Responder[JValue]): Handler =
- createHandler(responder, "text/json", (in: JValue) => pretty(render(in)))
-
- implicit def htmlResponderToHandler(responder: Responder[Seq[Node]]): Handler =
- createHandler(responder, "text/html", (in: Seq[Node]) => "<!DOCTYPE html>" + in.toString)
-
- implicit def textResponderToHandler(responder: Responder[String]): Handler =
- createHandler(responder, "text/plain")
-
- def createHandler[T <% AnyRef](responder: Responder[T], contentType: String,
- extractFn: T => String = (in: Any) => in.toString): Handler = {
- new AbstractHandler {
- def handle(target: String,
- baseRequest: Request,
- request: HttpServletRequest,
- response: HttpServletResponse) {
- response.setContentType("%s;charset=utf-8".format(contentType))
- response.setStatus(HttpServletResponse.SC_OK)
- baseRequest.setHandled(true)
- val result = responder(request)
- response.setHeader("Cache-Control", "no-cache, no-store, must-revalidate")
- response.getWriter().println(extractFn(result))
- }
- }
- }
-
- /** Creates a handler that always redirects the user to a given path */
- def createRedirectHandler(newPath: String): Handler = {
- new AbstractHandler {
- def handle(target: String,
- baseRequest: Request,
- request: HttpServletRequest,
- response: HttpServletResponse) {
- response.setStatus(302)
- response.setHeader("Location", baseRequest.getRootURL + newPath)
- baseRequest.setHandled(true)
- }
- }
- }
-
- /** Creates a handler for serving files from a static directory */
- def createStaticHandler(resourceBase: String): ResourceHandler = {
- val staticHandler = new ResourceHandler
- Option(getClass.getClassLoader.getResource(resourceBase)) match {
- case Some(res) =>
- staticHandler.setResourceBase(res.toString)
- case None =>
- throw new Exception("Could not find resource path for Web UI: " + resourceBase)
- }
- staticHandler
- }
-
- /**
- * Attempts to start a Jetty server at the supplied ip:port which uses the supplied handlers.
- *
- * If the desired port number is contented, continues incrementing ports until a free port is
- * found. Returns the chosen port and the jetty Server object.
- */
- def startJettyServer(ip: String, port: Int, handlers: Seq[(String, Handler)]): (Server, Int) = {
- val handlersToRegister = handlers.map { case(path, handler) =>
- val contextHandler = new ContextHandler(path)
- contextHandler.setHandler(handler)
- contextHandler.asInstanceOf[org.eclipse.jetty.server.Handler]
- }
-
- val handlerList = new HandlerList
- handlerList.setHandlers(handlersToRegister.toArray)
-
- @tailrec
- def connect(currentPort: Int): (Server, Int) = {
- val server = new Server(currentPort)
- val pool = new QueuedThreadPool
- pool.setDaemon(true)
- server.setThreadPool(pool)
- server.setHandler(handlerList)
-
- Try { server.start() } match {
- case s: Success[_] =>
- sys.addShutdownHook(server.stop()) // Be kind, un-bind
- (server, server.getConnectors.head.getLocalPort)
- case f: Failure[_] =>
- server.stop()
- logInfo("Failed to create UI at port, %s. Trying again.".format(currentPort))
- logInfo("Error was: " + f.toString)
- connect((currentPort + 1) % 65536)
- }
- }
-
- connect(port)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/ui/Page.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/ui/Page.scala b/core/src/main/scala/spark/ui/Page.scala
deleted file mode 100644
index 87376a1..0000000
--- a/core/src/main/scala/spark/ui/Page.scala
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * 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 spark.ui
-
-private[spark] object Page extends Enumeration {
- val Stages, Storage, Environment, Executors = Value
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/ui/SparkUI.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/ui/SparkUI.scala b/core/src/main/scala/spark/ui/SparkUI.scala
deleted file mode 100644
index 23ded44..0000000
--- a/core/src/main/scala/spark/ui/SparkUI.scala
+++ /dev/null
@@ -1,87 +0,0 @@
-/*
- * 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 spark.ui
-
-import javax.servlet.http.HttpServletRequest
-
-import org.eclipse.jetty.server.{Handler, Server}
-
-import spark.{Logging, SparkContext, SparkEnv, Utils}
-import spark.ui.env.EnvironmentUI
-import spark.ui.exec.ExecutorsUI
-import spark.ui.storage.BlockManagerUI
-import spark.ui.jobs.JobProgressUI
-import spark.ui.JettyUtils._
-
-/** Top level user interface for Spark */
-private[spark] class SparkUI(sc: SparkContext) extends Logging {
- val host = Option(System.getenv("SPARK_PUBLIC_DNS")).getOrElse(Utils.localHostName())
- val port = Option(System.getProperty("spark.ui.port")).getOrElse(SparkUI.DEFAULT_PORT).toInt
- var boundPort: Option[Int] = None
- var server: Option[Server] = None
-
- val handlers = Seq[(String, Handler)](
- ("/static", createStaticHandler(SparkUI.STATIC_RESOURCE_DIR)),
- ("/", createRedirectHandler("/stages"))
- )
- val storage = new BlockManagerUI(sc)
- val jobs = new JobProgressUI(sc)
- val env = new EnvironmentUI(sc)
- val exec = new ExecutorsUI(sc)
-
- // Add MetricsServlet handlers by default
- val metricsServletHandlers = SparkEnv.get.metricsSystem.getServletHandlers
-
- val allHandlers = storage.getHandlers ++ jobs.getHandlers ++ env.getHandlers ++
- exec.getHandlers ++ metricsServletHandlers ++ handlers
-
- /** Bind the HTTP server which backs this web interface */
- def bind() {
- try {
- val (srv, usedPort) = JettyUtils.startJettyServer("0.0.0.0", port, allHandlers)
- logInfo("Started Spark Web UI at http://%s:%d".format(host, usedPort))
- server = Some(srv)
- boundPort = Some(usedPort)
- } catch {
- case e: Exception =>
- logError("Failed to create Spark JettyUtils", e)
- System.exit(1)
- }
- }
-
- /** Initialize all components of the server */
- def start() {
- // NOTE: This is decoupled from bind() because of the following dependency cycle:
- // DAGScheduler() requires that the port of this server is known
- // This server must register all handlers, including JobProgressUI, before binding
- // JobProgressUI registers a listener with SparkContext, which requires sc to initialize
- jobs.start()
- exec.start()
- }
-
- def stop() {
- server.foreach(_.stop())
- }
-
- private[spark] def appUIAddress = "http://" + host + ":" + boundPort.getOrElse("-1")
-}
-
-private[spark] object SparkUI {
- val DEFAULT_PORT = "3030"
- val STATIC_RESOURCE_DIR = "spark/ui/static"
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/ui/UIUtils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/ui/UIUtils.scala b/core/src/main/scala/spark/ui/UIUtils.scala
deleted file mode 100644
index 51bb18d..0000000
--- a/core/src/main/scala/spark/ui/UIUtils.scala
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * 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 spark.ui
-
-import scala.xml.Node
-
-import spark.SparkContext
-
-/** Utility functions for generating XML pages with spark content. */
-private[spark] object UIUtils {
- import Page._
-
- /** Returns a spark page with correctly formatted headers */
- def headerSparkPage(content: => Seq[Node], sc: SparkContext, title: String, page: Page.Value)
- : Seq[Node] = {
- val jobs = page match {
- case Stages => <li class="active"><a href="/stages">Stages</a></li>
- case _ => <li><a href="/stages">Stages</a></li>
- }
- val storage = page match {
- case Storage => <li class="active"><a href="/storage">Storage</a></li>
- case _ => <li><a href="/storage">Storage</a></li>
- }
- val environment = page match {
- case Environment => <li class="active"><a href="/environment">Environment</a></li>
- case _ => <li><a href="/environment">Environment</a></li>
- }
- val executors = page match {
- case Executors => <li class="active"><a href="/executors">Executors</a></li>
- case _ => <li><a href="/executors">Executors</a></li>
- }
-
- <html>
- <head>
- <meta http-equiv="Content-type" content="text/html; charset=utf-8" />
- <link rel="stylesheet" href="/static/bootstrap.min.css" type="text/css" />
- <link rel="stylesheet" href="/static/webui.css" type="text/css" />
- <script src="/static/sorttable.js"></script>
- <title>{sc.appName} - {title}</title>
- </head>
- <body>
- <div class="navbar navbar-static-top">
- <div class="navbar-inner">
- <a href="/" class="brand"><img src="/static/spark-logo-77x50px-hd.png" /></a>
- <ul class="nav">
- {jobs}
- {storage}
- {environment}
- {executors}
- </ul>
- <p class="navbar-text pull-right"><strong>{sc.appName}</strong> application UI</p>
- </div>
- </div>
-
- <div class="container-fluid">
- <div class="row-fluid">
- <div class="span12">
- <h3 style="vertical-align: bottom; display: inline-block;">
- {title}
- </h3>
- </div>
- </div>
- {content}
- </div>
- </body>
- </html>
- }
-
- /** Returns a page with the spark css/js and a simple format. Used for scheduler UI. */
- def basicSparkPage(content: => Seq[Node], title: String): Seq[Node] = {
- <html>
- <head>
- <meta http-equiv="Content-type" content="text/html; charset=utf-8" />
- <link rel="stylesheet" href="/static/bootstrap.min.css" type="text/css" />
- <link rel="stylesheet" href="/static/webui.css" type="text/css" />
- <script src="/static/sorttable.js"></script>
- <title>{title}</title>
- </head>
- <body>
- <div class="container-fluid">
- <div class="row-fluid">
- <div class="span12">
- <h3 style="vertical-align: middle; display: inline-block;">
- <img src="/static/spark-logo-77x50px-hd.png" style="margin-right: 15px;" />
- {title}
- </h3>
- </div>
- </div>
- {content}
- </div>
- </body>
- </html>
- }
-
- /** Returns an HTML table constructed by generating a row for each object in a sequence. */
- def listingTable[T](
- headers: Seq[String],
- makeRow: T => Seq[Node],
- rows: Seq[T],
- fixedWidth: Boolean = false): Seq[Node] = {
-
- val colWidth = 100.toDouble / headers.size
- val colWidthAttr = if (fixedWidth) colWidth + "%" else ""
- var tableClass = "table table-bordered table-striped table-condensed sortable"
- if (fixedWidth) {
- tableClass += " table-fixed"
- }
-
- <table class={tableClass}>
- <thead>{headers.map(h => <th width={colWidthAttr}>{h}</th>)}</thead>
- <tbody>
- {rows.map(r => makeRow(r))}
- </tbody>
- </table>
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/ui/UIWorkloadGenerator.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/ui/UIWorkloadGenerator.scala b/core/src/main/scala/spark/ui/UIWorkloadGenerator.scala
deleted file mode 100644
index 5ff0572..0000000
--- a/core/src/main/scala/spark/ui/UIWorkloadGenerator.scala
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * 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 spark.ui
-
-import scala.util.Random
-
-import spark.SparkContext
-import spark.SparkContext._
-import spark.scheduler.cluster.SchedulingMode
-
-
-/**
- * Continuously generates jobs that expose various features of the WebUI (internal testing tool).
- *
- * Usage: ./run spark.ui.UIWorkloadGenerator [master]
- */
-private[spark] object UIWorkloadGenerator {
- val NUM_PARTITIONS = 100
- val INTER_JOB_WAIT_MS = 5000
-
- def main(args: Array[String]) {
- if (args.length < 2) {
- println("usage: ./spark-class spark.ui.UIWorkloadGenerator [master] [FIFO|FAIR]")
- System.exit(1)
- }
- val master = args(0)
- val schedulingMode = SchedulingMode.withName(args(1))
- val appName = "Spark UI Tester"
-
- if (schedulingMode == SchedulingMode.FAIR) {
- System.setProperty("spark.cluster.schedulingmode", "FAIR")
- }
- val sc = new SparkContext(master, appName)
-
- def setProperties(s: String) = {
- if(schedulingMode == SchedulingMode.FAIR) {
- sc.setLocalProperty("spark.scheduler.cluster.fair.pool", s)
- }
- sc.setLocalProperty(SparkContext.SPARK_JOB_DESCRIPTION, s)
- }
-
- val baseData = sc.makeRDD(1 to NUM_PARTITIONS * 10, NUM_PARTITIONS)
- def nextFloat() = (new Random()).nextFloat()
-
- val jobs = Seq[(String, () => Long)](
- ("Count", baseData.count),
- ("Cache and Count", baseData.map(x => x).cache.count),
- ("Single Shuffle", baseData.map(x => (x % 10, x)).reduceByKey(_ + _).count),
- ("Entirely failed phase", baseData.map(x => throw new Exception).count),
- ("Partially failed phase", {
- baseData.map{x =>
- val probFailure = (4.0 / NUM_PARTITIONS)
- if (nextFloat() < probFailure) {
- throw new Exception("This is a task failure")
- }
- 1
- }.count
- }),
- ("Partially failed phase (longer tasks)", {
- baseData.map{x =>
- val probFailure = (4.0 / NUM_PARTITIONS)
- if (nextFloat() < probFailure) {
- Thread.sleep(100)
- throw new Exception("This is a task failure")
- }
- 1
- }.count
- }),
- ("Job with delays", baseData.map(x => Thread.sleep(100)).count)
- )
-
- while (true) {
- for ((desc, job) <- jobs) {
- new Thread {
- override def run() {
- try {
- setProperties(desc)
- job()
- println("Job funished: " + desc)
- } catch {
- case e: Exception =>
- println("Job Failed: " + desc)
- }
- }
- }.start
- Thread.sleep(INTER_JOB_WAIT_MS)
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/ui/env/EnvironmentUI.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/ui/env/EnvironmentUI.scala b/core/src/main/scala/spark/ui/env/EnvironmentUI.scala
deleted file mode 100644
index b1be1a2..0000000
--- a/core/src/main/scala/spark/ui/env/EnvironmentUI.scala
+++ /dev/null
@@ -1,91 +0,0 @@
-/*
- * 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 spark.ui.env
-
-import javax.servlet.http.HttpServletRequest
-
-import scala.collection.JavaConversions._
-import scala.util.Properties
-import scala.xml.Node
-
-import org.eclipse.jetty.server.Handler
-
-import spark.ui.JettyUtils._
-import spark.ui.UIUtils
-import spark.ui.Page.Environment
-import spark.SparkContext
-
-
-private[spark] class EnvironmentUI(sc: SparkContext) {
-
- def getHandlers = Seq[(String, Handler)](
- ("/environment", (request: HttpServletRequest) => envDetails(request))
- )
-
- def envDetails(request: HttpServletRequest): Seq[Node] = {
- val jvmInformation = Seq(
- ("Java Version", "%s (%s)".format(Properties.javaVersion, Properties.javaVendor)),
- ("Java Home", Properties.javaHome),
- ("Scala Version", Properties.versionString),
- ("Scala Home", Properties.scalaHome)
- ).sorted
- def jvmRow(kv: (String, String)) = <tr><td>{kv._1}</td><td>{kv._2}</td></tr>
- def jvmTable =
- UIUtils.listingTable(Seq("Name", "Value"), jvmRow, jvmInformation, fixedWidth = true)
-
- val properties = System.getProperties.iterator.toSeq
- val classPathProperty = properties.find { case (k, v) =>
- k.contains("java.class.path")
- }.getOrElse(("", ""))
- val sparkProperties = properties.filter(_._1.startsWith("spark")).sorted
- val otherProperties = properties.diff(sparkProperties :+ classPathProperty).sorted
-
- val propertyHeaders = Seq("Name", "Value")
- def propertyRow(kv: (String, String)) = <tr><td>{kv._1}</td><td>{kv._2}</td></tr>
- val sparkPropertyTable =
- UIUtils.listingTable(propertyHeaders, propertyRow, sparkProperties, fixedWidth = true)
- val otherPropertyTable =
- UIUtils.listingTable(propertyHeaders, propertyRow, otherProperties, fixedWidth = true)
-
- val classPathEntries = classPathProperty._2
- .split(System.getProperty("path.separator", ":"))
- .filterNot(e => e.isEmpty)
- .map(e => (e, "System Classpath"))
- val addedJars = sc.addedJars.iterator.toSeq.map{case (path, time) => (path, "Added By User")}
- val addedFiles = sc.addedFiles.iterator.toSeq.map{case (path, time) => (path, "Added By User")}
- val classPath = (addedJars ++ addedFiles ++ classPathEntries).sorted
-
- val classPathHeaders = Seq("Resource", "Source")
- def classPathRow(data: (String, String)) = <tr><td>{data._1}</td><td>{data._2}</td></tr>
- val classPathTable =
- UIUtils.listingTable(classPathHeaders, classPathRow, classPath, fixedWidth = true)
-
- val content =
- <span>
- <h4>Runtime Information</h4> {jvmTable}
- <h4>Spark Properties</h4>
- {sparkPropertyTable}
- <h4>System Properties</h4>
- {otherPropertyTable}
- <h4>Classpath Entries</h4>
- {classPathTable}
- </span>
-
- UIUtils.headerSparkPage(content, sc, "Environment", Environment)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/ui/exec/ExecutorsUI.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/ui/exec/ExecutorsUI.scala b/core/src/main/scala/spark/ui/exec/ExecutorsUI.scala
deleted file mode 100644
index 0a7021f..0000000
--- a/core/src/main/scala/spark/ui/exec/ExecutorsUI.scala
+++ /dev/null
@@ -1,136 +0,0 @@
-package spark.ui.exec
-
-import javax.servlet.http.HttpServletRequest
-
-import scala.collection.mutable.{HashMap, HashSet}
-import scala.xml.Node
-
-import org.eclipse.jetty.server.Handler
-
-import spark.{ExceptionFailure, Logging, Utils, SparkContext}
-import spark.executor.TaskMetrics
-import spark.scheduler.cluster.TaskInfo
-import spark.scheduler.{SparkListenerTaskStart, SparkListenerTaskEnd, SparkListener}
-import spark.ui.JettyUtils._
-import spark.ui.Page.Executors
-import spark.ui.UIUtils
-
-
-private[spark] class ExecutorsUI(val sc: SparkContext) {
-
- private var _listener: Option[ExecutorsListener] = None
- def listener = _listener.get
-
- def start() {
- _listener = Some(new ExecutorsListener)
- sc.addSparkListener(listener)
- }
-
- def getHandlers = Seq[(String, Handler)](
- ("/executors", (request: HttpServletRequest) => render(request))
- )
-
- def render(request: HttpServletRequest): Seq[Node] = {
- val storageStatusList = sc.getExecutorStorageStatus
-
- val maxMem = storageStatusList.map(_.maxMem).fold(0L)(_+_)
- val memUsed = storageStatusList.map(_.memUsed()).fold(0L)(_+_)
- val diskSpaceUsed = storageStatusList.flatMap(_.blocks.values.map(_.diskSize)).fold(0L)(_+_)
-
- val execHead = Seq("Executor ID", "Address", "RDD blocks", "Memory used", "Disk used",
- "Active tasks", "Failed tasks", "Complete tasks", "Total tasks")
-
- def execRow(kv: Seq[String]) = {
- <tr>
- <td>{kv(0)}</td>
- <td>{kv(1)}</td>
- <td>{kv(2)}</td>
- <td sorttable_customkey={kv(3)}>
- {Utils.bytesToString(kv(3).toLong)} / {Utils.bytesToString(kv(4).toLong)}
- </td>
- <td sorttable_customkey={kv(5)}>
- {Utils.bytesToString(kv(5).toLong)}
- </td>
- <td>{kv(6)}</td>
- <td>{kv(7)}</td>
- <td>{kv(8)}</td>
- <td>{kv(9)}</td>
- </tr>
- }
-
- val execInfo = for (b <- 0 until storageStatusList.size) yield getExecInfo(b)
- val execTable = UIUtils.listingTable(execHead, execRow, execInfo)
-
- val content =
- <div class="row-fluid">
- <div class="span12">
- <ul class="unstyled">
- <li><strong>Memory:</strong>
- {Utils.bytesToString(memUsed)} Used
- ({Utils.bytesToString(maxMem)} Total) </li>
- <li><strong>Disk:</strong> {Utils.bytesToString(diskSpaceUsed)} Used </li>
- </ul>
- </div>
- </div>
- <div class = "row">
- <div class="span12">
- {execTable}
- </div>
- </div>;
-
- UIUtils.headerSparkPage(content, sc, "Executors (" + execInfo.size + ")", Executors)
- }
-
- def getExecInfo(a: Int): Seq[String] = {
- val execId = sc.getExecutorStorageStatus(a).blockManagerId.executorId
- val hostPort = sc.getExecutorStorageStatus(a).blockManagerId.hostPort
- val rddBlocks = sc.getExecutorStorageStatus(a).blocks.size.toString
- val memUsed = sc.getExecutorStorageStatus(a).memUsed().toString
- val maxMem = sc.getExecutorStorageStatus(a).maxMem.toString
- val diskUsed = sc.getExecutorStorageStatus(a).diskUsed().toString
- val activeTasks = listener.executorToTasksActive.get(a.toString).map(l => l.size).getOrElse(0)
- val failedTasks = listener.executorToTasksFailed.getOrElse(a.toString, 0)
- val completedTasks = listener.executorToTasksComplete.getOrElse(a.toString, 0)
- val totalTasks = activeTasks + failedTasks + completedTasks
-
- Seq(
- execId,
- hostPort,
- rddBlocks,
- memUsed,
- maxMem,
- diskUsed,
- activeTasks.toString,
- failedTasks.toString,
- completedTasks.toString,
- totalTasks.toString
- )
- }
-
- private[spark] class ExecutorsListener extends SparkListener with Logging {
- val executorToTasksActive = HashMap[String, HashSet[TaskInfo]]()
- val executorToTasksComplete = HashMap[String, Int]()
- val executorToTasksFailed = HashMap[String, Int]()
-
- override def onTaskStart(taskStart: SparkListenerTaskStart) {
- val eid = taskStart.taskInfo.executorId
- val activeTasks = executorToTasksActive.getOrElseUpdate(eid, new HashSet[TaskInfo]())
- activeTasks += taskStart.taskInfo
- }
-
- override def onTaskEnd(taskEnd: SparkListenerTaskEnd) {
- val eid = taskEnd.taskInfo.executorId
- val activeTasks = executorToTasksActive.getOrElseUpdate(eid, new HashSet[TaskInfo]())
- activeTasks -= taskEnd.taskInfo
- val (failureInfo, metrics): (Option[ExceptionFailure], Option[TaskMetrics]) =
- taskEnd.reason match {
- case e: ExceptionFailure =>
- executorToTasksFailed(eid) = executorToTasksFailed.getOrElse(eid, 0) + 1
- (Some(e), e.metrics)
- case _ =>
- executorToTasksComplete(eid) = executorToTasksComplete.getOrElse(eid, 0) + 1
- (None, Option(taskEnd.taskMetrics))
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/46eecd11/core/src/main/scala/spark/ui/jobs/IndexPage.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/spark/ui/jobs/IndexPage.scala b/core/src/main/scala/spark/ui/jobs/IndexPage.scala
deleted file mode 100644
index 8867a6c..0000000
--- a/core/src/main/scala/spark/ui/jobs/IndexPage.scala
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * 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 spark.ui.jobs
-
-import javax.servlet.http.HttpServletRequest
-
-import scala.xml.{NodeSeq, Node}
-
-import spark.scheduler.cluster.SchedulingMode
-import spark.ui.Page._
-import spark.ui.UIUtils._
-
-
-/** Page showing list of all ongoing and recently finished stages and pools*/
-private[spark] class IndexPage(parent: JobProgressUI) {
- def listener = parent.listener
-
- def render(request: HttpServletRequest): Seq[Node] = {
- listener.synchronized {
- val activeStages = listener.activeStages.toSeq
- val completedStages = listener.completedStages.reverse.toSeq
- val failedStages = listener.failedStages.reverse.toSeq
- val now = System.currentTimeMillis()
-
- var activeTime = 0L
- for (tasks <- listener.stageToTasksActive.values; t <- tasks) {
- activeTime += t.timeRunning(now)
- }
-
- val activeStagesTable = new StageTable(activeStages.sortBy(_.submissionTime).reverse, parent)
- val completedStagesTable = new StageTable(completedStages.sortBy(_.submissionTime).reverse, parent)
- val failedStagesTable = new StageTable(failedStages.sortBy(_.submissionTime).reverse, parent)
-
- val pools = listener.sc.getAllPools
- val poolTable = new PoolTable(pools, listener)
- val summary: NodeSeq =
- <div>
- <ul class="unstyled">
- <li>
- <strong>Total Duration: </strong>
- {parent.formatDuration(now - listener.sc.startTime)}
- </li>
- <li><strong>Scheduling Mode:</strong> {parent.sc.getSchedulingMode}</li>
- <li>
- <a href="#active"><strong>Active Stages:</strong></a>
- {activeStages.size}
- </li>
- <li>
- <a href="#completed"><strong>Completed Stages:</strong></a>
- {completedStages.size}
- </li>
- <li>
- <a href="#failed"><strong>Failed Stages:</strong></a>
- {failedStages.size}
- </li>
- </ul>
- </div>
-
- val content = summary ++
- {if (listener.sc.getSchedulingMode == SchedulingMode.FAIR) {
- <h4>{pools.size} Fair Scheduler Pools</h4> ++ poolTable.toNodeSeq
- } else {
- Seq()
- }} ++
- <h4 id="active">Active Stages ({activeStages.size})</h4> ++
- activeStagesTable.toNodeSeq++
- <h4 id="completed">Completed Stages ({completedStages.size})</h4> ++
- completedStagesTable.toNodeSeq++
- <h4 id ="failed">Failed Stages ({failedStages.size})</h4> ++
- failedStagesTable.toNodeSeq
-
- headerSparkPage(content, parent.sc, "Spark Stages", Stages)
- }
- }
-}