Skip to content

Commit b14ede7

Browse files
kayousterhoutmateiz
authored andcommitted
Remove broken/unused Connection.getChunkFIFO method.
This method appears to be broken -- since it never removes anything from messages, and it adds new messages to it, the while loop is an infinite loop. The method also does not appear to have ever been used since the code was added in 2012, so this commit removes it. cc @mateiz who originally added this method in case there's a reason it should be here! (63051dd) Author: Kay Ousterhout <kayousterhout@gmail.com> Closes #69 from kayousterhout/remove_get_fifo and squashes the following commits: 053bc59 [Kay Ousterhout] Remove broken/unused Connection.getChunkFIFO method.
1 parent f5ae38a commit b14ede7

1 file changed

Lines changed: 2 additions & 34 deletions

File tree

core/src/main/scala/org/apache/spark/network/Connection.scala

Lines changed: 2 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -171,7 +171,7 @@ class SendingConnection(val address: InetSocketAddress, selector_ : Selector,
171171
remoteId_ : ConnectionManagerId)
172172
extends Connection(SocketChannel.open, selector_, remoteId_) {
173173

174-
private class Outbox(fair: Int = 0) {
174+
private class Outbox {
175175
val messages = new Queue[Message]()
176176
val defaultChunkSize = 65536 //32768 //16384
177177
var nextMessageToBeUsed = 0
@@ -186,38 +186,6 @@ class SendingConnection(val address: InetSocketAddress, selector_ : Selector,
186186
}
187187

188188
def getChunk(): Option[MessageChunk] = {
189-
fair match {
190-
case 0 => getChunkFIFO()
191-
case 1 => getChunkRR()
192-
case _ => throw new Exception("Unexpected fairness policy in outbox")
193-
}
194-
}
195-
196-
private def getChunkFIFO(): Option[MessageChunk] = {
197-
/*logInfo("Using FIFO")*/
198-
messages.synchronized {
199-
while (!messages.isEmpty) {
200-
val message = messages(0)
201-
val chunk = message.getChunkForSending(defaultChunkSize)
202-
if (chunk.isDefined) {
203-
messages += message // this is probably incorrect, it wont work as fifo
204-
if (!message.started) {
205-
logDebug("Starting to send [" + message + "]")
206-
message.started = true
207-
message.startTime = System.currentTimeMillis
208-
}
209-
return chunk
210-
} else {
211-
message.finishTime = System.currentTimeMillis
212-
logDebug("Finished sending [" + message + "] to [" + getRemoteConnectionManagerId() +
213-
"] in " + message.timeTaken )
214-
}
215-
}
216-
}
217-
None
218-
}
219-
220-
private def getChunkRR(): Option[MessageChunk] = {
221189
messages.synchronized {
222190
while (!messages.isEmpty) {
223191
/*nextMessageToBeUsed = nextMessageToBeUsed % messages.size */
@@ -249,7 +217,7 @@ class SendingConnection(val address: InetSocketAddress, selector_ : Selector,
249217

250218
// outbox is used as a lock - ensure that it is always used as a leaf (since methods which
251219
// lock it are invoked in context of other locks)
252-
private val outbox = new Outbox(1)
220+
private val outbox = new Outbox()
253221
/*
254222
This is orthogonal to whether we have pending bytes to write or not - and satisfies a slightly
255223
different purpose. This flag is to see if we need to force reregister for write even when we

0 commit comments

Comments
 (0)