Alternate form of BufferedInputStream

Question

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
  }
}

Answer 1

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
  }
}

Answer 2

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

Answer 3

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.

Answer 4

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.