ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Refactoring...

Browse source
This commit is contained in:
Mosharaf Chowdhury 2010-10-28 15:09:58 -07:00
parent 681dba1373
commit 03d73dea9f
1 changed files with 14 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
src/scala/spark/Broadcast.scala
View file

@ -286,8 +286,19 @@ extends BroadcastRecipe with Logging {
}
listOfSources = listOfSources + newSourceInfo
}
}
}
private def addToListOfSources (newSourceInfos: ListBuffer[SourceInfo]) = {
listOfSources.synchronized {
newSourceInfos.foreach { srcInfo =>
if (listOfSources.contains(srcInfo)) {
listOfSources = listOfSources - srcInfo
}
listOfSources = listOfSources + srcInfo
}
}
}
class TalkToGuide (gInfo: SourceInfo)
extends Thread with Logging {
override def run = {
@ -308,21 +319,10 @@ extends BroadcastRecipe with Logging {
// Receive source information from Guide
var suitableSources =
oisGuide.readObject.asInstanceOf[ListBuffer[SourceInfo]]
oisGuide.readObject.asInstanceOf[ListBuffer[SourceInfo]]
logInfo("Received suitableSources from Master " + suitableSources)
// Update local list of known sources by adding or replacing
listOfSources.synchronized {
suitableSources.foreach { srcInfo =>
// Removing old copy of srcInfo to be replaced with a new one
// It works because case clases are compared by constructor params
if (listOfSources.contains(srcInfo)) {
listOfSources = listOfSources - srcInfo
}
listOfSources = listOfSources + srcInfo
}
}
addToListOfSources (suitableSources)
oisGuide.close
oosGuide.close