Skip to content

Commit cb9c8a9

Browse files
JoshRosenJoshRosen
committed
Extract BlockInfo classes from BlockManager.
This saves space, since the inner classes needed to keep a reference to the enclosing BlockManager.
1 parent 846b1cf commit cb9c8a9

File tree

2 files changed

+97
-75
lines changed

2 files changed

+97
-75
lines changed

‎core/src/main/scala/org/apache/spark/storage/BlockInfo.scala‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.spark.storage
19+
20+
import java.util.concurrent.ConcurrentHashMap
21+
22+
private[storage] trait BlockInfo {
23+
def level: StorageLevel
24+
def tellMaster: Boolean
25+
// To save space, 'pending' and 'failed' are encoded as special sizes:
26+
@volatile var size: Long = BlockInfo.BLOCK_PENDING
27+
private def pending: Boolean = size == BlockInfo.BLOCK_PENDING
28+
private def failed: Boolean = size == BlockInfo.BLOCK_FAILED
29+
private def initThread: Thread = BlockInfo.blockInfoInitThreads.get(this)
30+
31+
setInitThread()
32+
33+
private def setInitThread() {
34+
// Set current thread as init thread - waitForReady will not block this thread
35+
// (in case there is non trivial initialization which ends up calling waitForReady as part of
36+
// initialization itself)
37+
BlockInfo.blockInfoInitThreads.put(this, Thread.currentThread())
38+
}
39+
40+
/**
41+
* Wait for this BlockInfo to be marked as ready (i.e. block is finished writing).
42+
* Return true if the block is available, false otherwise.
43+
*/
44+
def waitForReady(): Boolean = {
45+
if (pending && initThread != Thread.currentThread()) {
46+
synchronized {
47+
while (pending) this.wait()
48+
}
49+
}
50+
!failed
51+
}
52+
53+
/** Mark this BlockInfo as ready (i.e. block is finished writing) */
54+
def markReady(sizeInBytes: Long) {
55+
require (sizeInBytes >= 0, "sizeInBytes was negative: " + sizeInBytes)
56+
assert (pending)
57+
size = sizeInBytes
58+
BlockInfo.blockInfoInitThreads.remove(this)
59+
synchronized {
60+
this.notifyAll()
61+
}
62+
}
63+
64+
/** Mark this BlockInfo as ready but failed */
65+
def markFailure() {
66+
assert (pending)
67+
size = BlockInfo.BLOCK_FAILED
68+
BlockInfo.blockInfoInitThreads.remove(this)
69+
synchronized {
70+
this.notifyAll()
71+
}
72+
}
73+
}
74+
75+
private object BlockInfo {
76+
// initThread is logically a BlockInfo field, but we store it here because
77+
// it's only needed while this block is in the 'pending' state and we want
78+
// to minimize BlockInfo's memory footprint.
79+
private val blockInfoInitThreads = new ConcurrentHashMap[BlockInfo, Thread]
80+
81+
private val BLOCK_PENDING: Long = -1L
82+
private val BLOCK_FAILED: Long = -2L
83+
}
84+
85+
// All shuffle blocks have the same `level` and `tellMaster` properties,
86+
// so we can save space by not storing them in each instance:
87+
private[storage] class ShuffleBlockInfo extends BlockInfo {
88+
// These need to be defined using 'def' instead of 'val' in order for
89+
// the compiler to eliminate the fields:
90+
def level: StorageLevel = StorageLevel.DISK_ONLY
91+
def tellMaster: Boolean = false
92+
}
93+
94+
private[storage] class BlockInfoImpl(val level: StorageLevel, val tellMaster: Boolean)
95+
extends BlockInfo {
96+
// Intentionally left blank
97+
}

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

Lines changed: 0 additions & 75 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@ package org.apache.spark.storage
1919

2020
import java.io.{InputStream, OutputStream}
2121
import java.nio.{ByteBuffer, MappedByteBuffer}
22-
import java.util.concurrent.ConcurrentHashMap
2322

2423
import scala.collection.mutable.{HashMap, ArrayBuffer}
2524
import scala.util.Random
@@ -47,80 +46,6 @@ private[spark] class BlockManager(
4746
maxMemory: Long)
4847
extends Logging {
4948

50-
51-
// initThread is logically a BlockInfo field, but we store it here because
52-
// it's only needed while this block is in the 'pending' state and we want
53-
// to minimize BlockInfo's memory footprint.
54-
private val blockInfoInitThreads = new ConcurrentHashMap[BlockInfo, Thread]
55-
56-
private val BLOCK_PENDING: Long = -1L
57-
private val BLOCK_FAILED: Long = -2L
58-
59-
private trait BlockInfo {
60-
def level: StorageLevel
61-
def tellMaster: Boolean
62-
@volatile var size: Long = BLOCK_PENDING // also encodes 'pending' and 'failed' to save space
63-
private def pending: Boolean = size == BLOCK_PENDING
64-
private def failed: Boolean = size == BLOCK_FAILED
65-
private def initThread: Thread = blockInfoInitThreads.get(this)
66-
67-
setInitThread()
68-
69-
private def setInitThread() {
70-
// Set current thread as init thread - waitForReady will not block this thread
71-
// (in case there is non trivial initialization which ends up calling waitForReady as part of
72-
// initialization itself)
73-
blockInfoInitThreads.put(this, Thread.currentThread())
74-
}
75-
76-
/**
77-
* Wait for this BlockInfo to be marked as ready (i.e. block is finished writing).
78-
* Return true if the block is available, false otherwise.
79-
*/
80-
def waitForReady(): Boolean = {
81-
if (pending && initThread != Thread.currentThread()) {
82-
synchronized {
83-
while (pending) this.wait()
84-
}
85-
}
86-
!failed
87-
}
88-
89-
/** Mark this BlockInfo as ready (i.e. block is finished writing) */
90-
def markReady(sizeInBytes: Long) {
91-
require (sizeInBytes >= 0, "sizeInBytes was negative: " + sizeInBytes)
92-
assert (pending)
93-
size = sizeInBytes
94-
blockInfoInitThreads.remove(this)
95-
synchronized {
96-
this.notifyAll()
97-
}
98-
}
99-
100-
/** Mark this BlockInfo as ready but failed */
101-
def markFailure() {
102-
assert (pending)
103-
size = BLOCK_FAILED
104-
blockInfoInitThreads.remove(this)
105-
synchronized {
106-
this.notifyAll()
107-
}
108-
}
109-
}
110-
111-
// All shuffle blocks have the same `level` and `tellMaster` properties,
112-
// so we can save space by not storing them in each instance:
113-
private class ShuffleBlockInfo extends BlockInfo {
114-
// These need to be defined using 'def' instead of 'val' in order for
115-
// the compiler to eliminate the fields:
116-
def level: StorageLevel = StorageLevel.DISK_ONLY
117-
def tellMaster: Boolean = false
118-
}
119-
120-
private class BlockInfoImpl(val level: StorageLevel, val tellMaster: Boolean) extends BlockInfo {
121-
// Intentionally left blank
122-
}
123-
12449
val shuffleBlockManager = new ShuffleBlockManager(this)
12550
val diskBlockManager = new DiskBlockManager(
12651
System.getProperty("spark.local.dir", System.getProperty("java.io.tmpdir")))

0 commit comments

Comments
 (0)