diff options
author | Matei Zaharia <matei@eecs.berkeley.edu> | 2012-10-08 16:54:28 -0700 |
---|---|---|
committer | Matei Zaharia <matei@eecs.berkeley.edu> | 2012-10-08 16:54:28 -0700 |
commit | 1231eb12e675fec47bc2d3139041b1c178a08c37 (patch) | |
tree | dba0835cf3d79faa876b30f013166d5ac99d06b0 /core/src/main | |
parent | 4f1e0325f1a9ab28c2e583e70e70d8f7bde127ff (diff) | |
parent | edc67bfba8b7875bb751f1a8c84af7135a1d3d39 (diff) | |
download | spark-1231eb12e675fec47bc2d3139041b1c178a08c37.tar.gz spark-1231eb12e675fec47bc2d3139041b1c178a08c37.tar.bz2 spark-1231eb12e675fec47bc2d3139041b1c178a08c37.zip |
Merge pull request #259 from mosharaf/bc-fix-dev
Synchronization bug fix in broadcast implementations
Diffstat (limited to 'core/src/main')
4 files changed, 28 insertions, 23 deletions
diff --git a/core/src/main/scala/spark/broadcast/BitTorrentBroadcast.scala b/core/src/main/scala/spark/broadcast/BitTorrentBroadcast.scala index cf20f456c4..ef27bbb502 100644 --- a/core/src/main/scala/spark/broadcast/BitTorrentBroadcast.scala +++ b/core/src/main/scala/spark/broadcast/BitTorrentBroadcast.scala @@ -311,9 +311,11 @@ private[spark] class BitTorrentBroadcast[T](@transient var value_ : T, isLocal: var threadPool = Utils.newDaemonFixedThreadPool(MultiTracker.MaxChatSlots) while (hasBlocks.get < totalBlocks) { - var numThreadsToCreate = - math.min(listOfSources.size, MultiTracker.MaxChatSlots) - + var numThreadsToCreate = 0 + listOfSources.synchronized { + numThreadsToCreate = math.min(listOfSources.size, MultiTracker.MaxChatSlots) - threadPool.getActiveCount + } while (hasBlocks.get < totalBlocks && numThreadsToCreate > 0) { var peerToTalkTo = pickPeerToTalkToRandom @@ -726,7 +728,6 @@ private[spark] class BitTorrentBroadcast[T](@transient var value_ : T, isLocal: guidePortLock.synchronized { guidePortLock.notifyAll() } try { - // Don't stop until there is a copy in HDFS while (!stopBroadcast) { var clientSocket: Socket = null try { @@ -734,14 +735,17 @@ private[spark] class BitTorrentBroadcast[T](@transient var value_ : T, isLocal: clientSocket = serverSocket.accept() } catch { case e: Exception => { - logError("GuideMultipleRequests Timeout.") - // Stop broadcast if at least one worker has connected and // everyone connected so far are done. Comparing with // listOfSources.size - 1, because it includes the Guide itself - if (listOfSources.size > 1 && - setOfCompletedSources.size == listOfSources.size - 1) { - stopBroadcast = true + listOfSources.synchronized { + setOfCompletedSources.synchronized { + if (listOfSources.size > 1 && + setOfCompletedSources.size == listOfSources.size - 1) { + stopBroadcast = true + logInfo("GuideMultipleRequests Timeout. stopBroadcast == true.") + } + } } } } @@ -922,9 +926,7 @@ private[spark] class BitTorrentBroadcast[T](@transient var value_ : T, isLocal: serverSocket.setSoTimeout(MultiTracker.ServerSocketTimeout) clientSocket = serverSocket.accept() } catch { - case e: Exception => { - logError("ServeMultipleRequests Timeout.") - } + case e: Exception => { } } if (clientSocket != null) { logDebug("Serve: Accepted new client connection:" + clientSocket) diff --git a/core/src/main/scala/spark/broadcast/MultiTracker.scala b/core/src/main/scala/spark/broadcast/MultiTracker.scala index dd8e6dd246..5e76dedb94 100644 --- a/core/src/main/scala/spark/broadcast/MultiTracker.scala +++ b/core/src/main/scala/spark/broadcast/MultiTracker.scala @@ -228,7 +228,7 @@ extends Logging { var oosTracker: ObjectOutputStream = null var oisTracker: ObjectInputStream = null - var gInfo: SourceInfo = SourceInfo("", SourceInfo.TxOverGoToDefault) + var gInfo: SourceInfo = SourceInfo("", SourceInfo.TxNotStartedRetry) var retriesLeft = MultiTracker.MaxRetryCount do { diff --git a/core/src/main/scala/spark/broadcast/SourceInfo.scala b/core/src/main/scala/spark/broadcast/SourceInfo.scala index 705dd6fd81..c79bb93c38 100644 --- a/core/src/main/scala/spark/broadcast/SourceInfo.scala +++ b/core/src/main/scala/spark/broadcast/SourceInfo.scala @@ -27,9 +27,10 @@ extends Comparable[SourceInfo] with Logging { * Helper Object of SourceInfo for its constants */ private[spark] object SourceInfo { - // Constants for special values of listenPort + // Broadcast has not started yet! Should never happen. val TxNotStartedRetry = -1 - val TxOverGoToDefault = 0 + // Broadcast has already finished. Try default mechanism. + val TxOverGoToDefault = -3 // Other constants val StopBroadcast = -2 val UnusedParam = 0 diff --git a/core/src/main/scala/spark/broadcast/TreeBroadcast.scala b/core/src/main/scala/spark/broadcast/TreeBroadcast.scala index 5bd40a40e3..fa676e9064 100644 --- a/core/src/main/scala/spark/broadcast/TreeBroadcast.scala +++ b/core/src/main/scala/spark/broadcast/TreeBroadcast.scala @@ -292,15 +292,17 @@ extends Broadcast[T](id) with Logging with Serializable { clientSocket = serverSocket.accept } catch { case e: Exception => { - logError("GuideMultipleRequests Timeout.") - // Stop broadcast if at least one worker has connected and - // everyone connected so far are done. - // Comparing with listOfSources.size - 1, because the Guide itself - // is included - if (listOfSources.size > 1 && - setOfCompletedSources.size == listOfSources.size - 1) { - stopBroadcast = true + // everyone connected so far are done. Comparing with + // listOfSources.size - 1, because it includes the Guide itself + listOfSources.synchronized { + setOfCompletedSources.synchronized { + if (listOfSources.size > 1 && + setOfCompletedSources.size == listOfSources.size - 1) { + stopBroadcast = true + logInfo("GuideMultipleRequests Timeout. stopBroadcast == true.") + } + } } } } @@ -492,7 +494,7 @@ extends Broadcast[T](id) with Logging with Serializable { serverSocket.setSoTimeout(MultiTracker.ServerSocketTimeout) clientSocket = serverSocket.accept } catch { - case e: Exception => logError("ServeMultipleRequests Timeout.") + case e: Exception => { } } if (clientSocket != null) { |