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

Bug fix: splitsInRequestBitVector(splitIndex) was wrongly set to false after receiving just one block in Blocked implementations that receive multiple blocks at a time.

Browse source
This commit is contained in:
Mosharaf Chowdhury 2011-01-03 19:05:52 -08:00
parent 07e778d7fa
commit 7eb334d97c
3 changed files with 6 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
src/scala/spark/CustomBlockedInMemoryShuffle.scala
View file

@ -361,11 +361,6 @@ extends Shuffle[K, V, C] with Logging {
hasSplits += 1
}
// We have received splitIndex
splitsInRequestBitVector.synchronized {
splitsInRequestBitVector.set(splitIndex, false)
}
// Consistent state in accounting variables
receptionSucceeded = true
@ -384,11 +379,8 @@ extends Shuffle[K, V, C] with Logging {
logInfo("ShuffleClient had a " + e)
}
} finally {
// If reception failed, unset for future retry
if (!receptionSucceeded) {
splitsInRequestBitVector.synchronized {
splitsInRequestBitVector.set(splitIndex, false)
}
splitsInRequestBitVector.synchronized {
splitsInRequestBitVector.set(splitIndex, false)
}
cleanUpConnections()
}

12
src/scala/spark/CustomBlockedLocalFileShuffle.scala
View file

@ -348,11 +348,6 @@ extends Shuffle[K, V, C] with Logging {
hasSplits += 1
}
// We have received splitIndex
splitsInRequestBitVector.synchronized {
splitsInRequestBitVector.set(splitIndex, false)
}
// Consistent state in accounting variables
receptionSucceeded = true
@ -371,11 +366,8 @@ extends Shuffle[K, V, C] with Logging {
logInfo("ShuffleClient had a " + e)
}
} finally {
// If reception failed, unset for future retry
if (!receptionSucceeded) {
splitsInRequestBitVector.synchronized {
splitsInRequestBitVector.set(splitIndex, false)
}
splitsInRequestBitVector.synchronized {
splitsInRequestBitVector.set(splitIndex, false)
}
cleanUpConnections()
}

12
src/scala/spark/TrackedCustomBlockedLocalFileShuffle.scala
View file

@ -548,11 +548,6 @@ extends Shuffle[K, V, C] with Logging {
hasSplits += 1
}
// We have received splitIndex
splitsInRequestBitVector.synchronized {
splitsInRequestBitVector.set(splitIndex, false)
}
receptionSucceeded = true
logInfo("END READ: " + requestPath)
@ -574,11 +569,8 @@ extends Shuffle[K, V, C] with Logging {
logInfo("ShuffleClient had a " + e)
}
} finally {
// If reception failed, unset for future retry
if (!receptionSucceeded) {
splitsInRequestBitVector.synchronized {
splitsInRequestBitVector.set(splitIndex, false)
}
splitsInRequestBitVector.synchronized {
splitsInRequestBitVector.set(splitIndex, false)
}
cleanUp()
}