11
\$\begingroup\$

I was having a problem with the TCP/IP socket communications in a web app where the server that the app was talking to would occasionally send bursts of data that would overflow the low level stream buffer, resulting in lost data.

The solution I came up with is basically an unthreaded form of the StreamGobbler and I have been calling it a GreedyBufferedInputStream. The basic idea is that just about any call to this class will include draining the source InputStream.

I'm still fairly new to Scala, so please let me know how this code could be improved. Also, is there anything in the code which could be done more efficiently? Performance is critically important.

package edu.stsci.util

import org.slf4j.Logger

import java.io.InputStream
import java.util

class GreedyBufferedInputStream extends InputStream {
  private var logger: Logger = null
  private var source: InputStream = null
  private val data: util.LinkedList[DataBlock] = new util.LinkedList[DataBlock]()
  private var currentBlock: DataBlock = null

  def this(logger: Logger, source: InputStream) {
    this()

    this.logger = logger
    this.source = source

    drainSource()
  }

  def this(source: InputStream) {
    this(null, source)
  }

  def read() = {
    prepareToRead()
    if (currentBlock == null) -1
    else currentBlock.read()
  }

  override def read(destination: Array[Byte], offset: Int, length: Int): Int = {
    prepareToRead()
    if (currentBlock == null) return -1

    var bytesRead = currentBlock.read(offset, length, destination)

    while (bytesRead < length) {
      prepareToRead()
      if (currentBlock == null) {// EOF
        return bytesRead
      }
      else {
        bytesRead += currentBlock.read((offset + bytesRead), (length - bytesRead), destination)
      }
    }

    bytesRead
  }

  override def read(destination: Array[Byte]) = read(destination, 0, destination.length)

  override def skip(length: Long): Long = {
    prepareToRead()
    if (currentBlock == null) return -1

    var bytesSkipped = currentBlock.skip(length)

    while (bytesSkipped < length) {
      prepareToRead()
      if (currentBlock == null) { // EOF
        return bytesSkipped
      }
      else {
        bytesSkipped += currentBlock.skip(length - bytesSkipped)
      }
    }

    bytesSkipped
  }

  override def close() {
    super.close()
  }

  override def available() = {
    drainSource()

    var result = 0

    if (currentBlock != null) result += currentBlock.available

    val it = data.iterator()
    while (it.hasNext) {
      val next = it.next()
      result += next.available
    }

    result
  }

  private def drainSource() {
    if (source == null) return // EOF

    if (source.available() > 0) {
      val raw = new Array[Byte](source.available())
      val length = source.read(raw)
      val block = new DataBlock(raw, length)
      data.add(block)
    }
  }

  private def prepareToRead() {
    drainSource()

    if (currentBlock != null) {
      val done = currentBlock.isDone
      if (done) currentBlock = null
      else return // we have a current block
    }

    if (data.isEmpty) { // no choice but to block
      if (source == null) return // have reached EOF

      val raw = new Array[Byte](1024)
      val length = source.read(raw)
      if (length < 0) {
        source = null
        return
      }
      currentBlock = new DataBlock(raw, length)

    }
    else currentBlock = data.remove()
  }
}

class DataBlock(data: Array[Byte], length: Int) {

  var readPos = -1

  def isDone: Boolean = (readPos >= length)

  def available: Int = {
    if (readPos < 0) length
    else (length - readPos)
  }

  def read(): Int = {
    if (readPos < 0) readPos = 0

    val raw = data(readPos)
    readPos += 1
    raw & 0xff
  }

  def read(offset: Int, length: Int, destination: Array[Byte]): Int = {
    if (readPos < 0) readPos = 0

    var readCount = {
      if (available >= length) length
      else available
    }

    Array.copy(data, readPos, destination, offset, readCount)
    readPos += readCount

    readCount
  }

  def skip(length: Long): Long = {
    if (readPos < 0) readPos = 0

    var readCount = {
      if (available >= length) length
      else available
    }

    readPos += readCount.toInt

    readCount
  }
}
\$\endgroup\$
1
  • \$\begingroup\$ Would it be at all helpful if I added the unit tests? \$\endgroup\$ Jul 17, 2012 at 14:30

4 Answers 4

3
+50
\$\begingroup\$

I think that the use of a LinkedList is not optimal. In every invocation of drainSource you potentially add a new DataBlock ant the end of the linked list; this has performance O(n) (length of the list). I propose to use a Vector instead of the LinkedList.

Another point (but this is not Scala specific): Why do you initialize the field readPos in class DataBlock with -1? You have many special cases due to this choice. I would implement class DataBlock as follows:

class DataBlock(data: Array[Byte], length: Int) {
  require(length > 0)

  private var readPos = 0

  def isDone: Boolean = readPos >= length

  def available: Int = length - readPos

  def read(): Int = try { data(readPos) & 0xff } finally { readPos += 1 }

  def read(offset: Int, length: Int, destination: Array[Byte]): Int = {
    val readCount = length min available
    Array.copy(data, readPos, destination, offset, readCount)
    readPos += readCount
    readCount
  }

  def skip(length: Long): Long = {
    val readCount = length min available
    readPos += readCount.toInt
    readCount
  }
}
\$\endgroup\$
2
  • \$\begingroup\$ I used java.util.LinkedList as I believe that it offers constant time adds at the end as well as constant time removes from the beginning. I dropped your DataBlock implementation in and it passed all the tests. \$\endgroup\$ Jul 18, 2012 at 17:53
  • \$\begingroup\$ @Donald.McLean you are right, missed that you use a Java list and not the Scala one. \$\endgroup\$
    – Dominik
    Jul 18, 2012 at 19:48
3
\$\begingroup\$

Just some random bits of feedback, not really addressing the code as a whole.

First, I would handle constructing the class as follows:

class GreedyBufferedInputStream(logger: Logger = null, initialSource: InputStream)
extends InputStream {

  private var source: InputStream = initialSource
  private val data: util.LinkedList[DataBlock] = new util.LinkedList()
  private var currentBlock: DataBlock = null

  drainSource()

This eliminates a var and both auxiliary constructors.

I would rewrite the available() method as follows:

  override def available() = {
    drainSource()
    import collection.JavaConverters._
    data.asScala.map(_.available).sum +
      Option(currentBlock).map(_.available).getOrElse(0)
  }

And I would replace this:

var readCount = {
  if (available >= length) length
  else available
}

with

val readCount = length min available
\$\endgroup\$
1
  • \$\begingroup\$ I like this answer, but I thought that Dominik's answer was slightly more useful. If I could, I would have split the bonus. \$\endgroup\$ Jul 20, 2012 at 11:53
0
\$\begingroup\$

To be precise I have no real solution to your problem, but if I think about it there should be a few possibilities to find one: First you could think about limiting data send by the App. A limitation could mean a two way communication and/or a sending buffer on side of the App. So you can regulate the data sent and maybe zip it to be more efficient etc. A second solution could be a pool of multiple stream buffers which can be chosen in case of a data burst to handle all incomming data. Because I am a Python programmer I cant tell you if your Scala code can be improved or not. I hope I could help a bit.

\$\endgroup\$
1
  • \$\begingroup\$ Thank you for your answer. However, I tried to be clear - the code I gave is the answer to my original problem and I am now looking for a review of this code. \$\endgroup\$ Jul 16, 2012 at 11:54
0
\$\begingroup\$

No time to read it in any detail, but as a general observation of scala style most scala users would be looking to make far more use of immutable ways of doing things. I use var very sparingly these days and never use null as an indicator of failure or empty something - I use Option[X] instead.

\$\endgroup\$
2
  • \$\begingroup\$ I actually understand Option - it's a startlingly useful construct. However, I fail to see, within the context of this class, where using Option would add any value. I could be wrong, but you'll have to provide some actual justification to convince me. \$\endgroup\$ Jul 18, 2012 at 14:44
  • \$\begingroup\$ I specifically said I have no time to look in detail at your classes. You may be right. Null creeps in from java - I try to wrap it up as soon as possible. \$\endgroup\$
    – Tim Pigden
    Jul 18, 2012 at 15:01

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.