From b1745b3103ece683807c7dcf16d8ce2bab3596cf Mon Sep 17 00:00:00 2001
From: Mosharaf Chowdhury <mosharaf@mosharaf-ubuntu.(none)>
Date: Sat, 4 Dec 2010 13:55:25 -0800
Subject: [PATCH] Removed an unnecessary byte array in the middle. Probabaly
will have to bring it back if we do block level data movement.
---
src/scala/spark/LocalFileShuffle.scala | 20 +++-----------------
1 file changed, 3 insertions(+), 17 deletions(-)
diff --git a/src/scala/spark/LocalFileShuffle.scala b/src/scala/spark/LocalFileShuffle.scala
index eb9905b77f..e3998b7774 100644
--- a/src/scala/spark/LocalFileShuffle.scala
+++ b/src/scala/spark/LocalFileShuffle.scala
@@ -146,8 +146,6 @@ class LocalFileShuffle[K, V, C] extends Shuffle[K, V, C] with Logging {
private var oosSource: ObjectOutputStream = null
private var oisSource: ObjectInputStream = null
- private var byteArray: Array[Byte] = null
-
private var receptionSucceeded = false
override def run: Unit = {
@@ -187,21 +185,9 @@ class LocalFileShuffle[K, V, C] extends Shuffle[K, V, C] with Logging {
// Receive the file
if (requestedFileLen != -1) {
- byteArray = new Array[Byte] (requestedFileLen)
- var bytesRead = isSource.read (byteArray, 0, byteArray.length)
- var alreadyRead = bytesRead
-
- while (alreadyRead < requestedFileLen) {
- bytesRead = isSource.read(byteArray, alreadyRead,
- (byteArray.length - alreadyRead))
- if(bytesRead > 0) {
- alreadyRead = alreadyRead + bytesRead
- }
- }
-
- // Now add this to combiners
- val inputStream = new ObjectInputStream (
- new ByteArrayInputStream(byteArray))
+ // Add this to combiners
+ val inputStream = new ObjectInputStream (isSource)
+
try{
while (true) {
val (k, c) = inputStream.readObject.asInstanceOf[(K, C)]
--
GitLab