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

Bug fixes to RateLimitedOutputStream

Browse source
This commit is contained in:
Matei Zaharia 2012-09-01 00:31:15 -07:00
parent 44758aa8e2
commit f84d2bbe55
2 changed files with 4 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
core/src/main/scala/spark/util/RateLimitedOutputStream.scala
View file

@ -21,7 +21,7 @@ class RateLimitedOutputStream(out: OutputStream, bytesPerSec: Int) extends Outpu
while (pos < length) {
val writeSize = math.min(length - pos, CHUNK_SIZE)
waitToWrite(writeSize)
out.write(bytes, offset + pos, length - pos)
out.write(bytes, offset + pos, writeSize)
pos += writeSize
}
}

4
streaming/src/main/scala/spark/streaming/util/RawTextSender.scala
View file

@ -15,6 +15,7 @@ object RawTextSender extends Logging {
def main(args: Array[String]) {
if (args.length != 4) {
System.err.println("Usage: RawTextSender <port> <file> <blockSize> <bytesPerSec>")
System.exit(1)
}
// Parse the arguments using a pattern match
val Array(IntParam(port), file, IntParam(blockSize), IntParam(bytesPerSec)) = args
@ -36,6 +37,7 @@ object RawTextSender extends Logging {
while (true) {
val socket = serverSocket.accept()
logInfo("Got a new connection")
val out = new RateLimitedOutputStream(socket.getOutputStream, bytesPerSec)
try {
while (true) {
@ -43,7 +45,7 @@ object RawTextSender extends Logging {
}
} catch {
case e: IOException =>
logError("Socket closed: ", e)
logError("Socket closed", e)
socket.close()
}
}