ByteBufferPerf.scala

/**
 * To compile:
 *  scalac -optimize ByteBufferPerf.scala
 *
 * JAVA_OPTS="-Xmx2g" scala IntArrayPerf 10
 *  49  62  48  45  48  45  48  50  47  45
 *
 * JAVA_OPTS="-Xmx2g" scala ByteBufferPerf 10
 *  479 491 484 480 484 481 477 477 472 473
 *
 * JAVA_OPTS="-Xmx2g" scala DirectByteBufferPerf 10
 *  910 321 316 320 323 317 320 316 319 323
 *
 * JAVA_OPTS="-Xmx2g" scala RamDiskByteBufferPerf 10
 *  985 320 293 292 291 298 292 294 299 292
 *
 * JAVA_OPTS="-Xmx2g" scala IntBufferPerf 10
 *  321 332 324 333 326 330 331 334 333 325
 *
 * JAVA_OPTS="-Xmx2g" scala DirectIntBufferPerf 10
 *  85  95  84  88  88  82  84  88  84  85
 *
 * JAVA_OPTS="-Xmx2g" scala RamDiskIntBufferPerf 10
 *  780 119 86  82  83  88  83  83  87  82
 *
 * JAVA_OPTS="-Xmx2g" scala DirectUnsafePerf 10
 *  82 91  84  84  79  84  77  78  83  84
 *
 * JAVA_OPTS="-Xmx2g" scala RamDiskUnsafePerf 10
 *  4815 110 77  77  77  77  76  77  76  77
 *
 * JAVA_OPTS="-Xmx2g" scala UnsafeByteBufferReaderPerf 10
 *  5979  83  52  47  53  48  48  53  47  47
 */

import java.nio.{ByteBuffer, ByteOrder, IntBuffer}

import scala.testing.Benchmark


object ByteBufferPerf extends Benchmark {

  val DATA_SIZE = 100 * 1000 * 1000 // 100 million

  val byteArray = new Array[Byte](4 * ByteBufferPerf.DATA_SIZE)
  var i = 0
  while (i < DATA_SIZE) {
    i += 1
    byteArray(i) = i.toByte
  }

  val data = java.nio.ByteBuffer.allocate(4 * ByteBufferPerf.DATA_SIZE)
  data.order(java.nio.ByteOrder.nativeOrder)
  data.put(byteArray)
  data.rewind()

  override def run = {
    var i = 0
    var sum = 0L
    while (i < ByteBufferPerf.DATA_SIZE) {
      sum += data.getInt
      i += 1
    }
    println("sum: " + sum)
  }

  override def setUp = data.rewind()
}


object DirectByteBufferPerf extends Benchmark {

  val byteArray = ByteBufferPerf.data
  val data = ByteBuffer.allocateDirect(4 * ByteBufferPerf.DATA_SIZE)
  data.order(ByteOrder.nativeOrder)
  data.put(byteArray)
  data.rewind()

  override def run = {
    var i = 0
    var sum = 0L
    while (i < ByteBufferPerf.DATA_SIZE) {
      sum += data.getInt
      i += 1
    }
    println("sum: " + sum)
  }

  override def setUp = data.rewind()
}


object RamDiskByteBufferPerf extends Benchmark {

  val channel = new java.io.RandomAccessFile("/Volumes/ramdisk/test.dat", "r").getChannel()
  val data: java.nio.MappedByteBuffer = channel.map(
    java.nio.channels.FileChannel.MapMode.READ_ONLY,
    0,
    math.min(channel.size(), java.lang.Integer.MAX_VALUE))

  data.order(java.nio.ByteOrder.nativeOrder)

  override def run = {
    var i = 0
    var sum = 0L
    while (i < ByteBufferPerf.DATA_SIZE) {
      sum += data.getInt
      i += 1
    }
    println("sum: " + sum)
  }

  override def setUp = data.rewind()
}


object IntBufferPerf extends Benchmark {

  val data = ByteBufferPerf.data.asIntBuffer()

  override def run = {
    var i = 0
    var sum = 0L
    while (i < ByteBufferPerf.DATA_SIZE) {
      sum += data.get
      i += 1
    }
    println("sum: " + sum)
  }

  override def setUp = data.rewind()
}


object DirectIntBufferPerf extends Benchmark {

  val data = DirectByteBufferPerf.data.asIntBuffer()

  override def run = {
    var i = 0
    var sum = 0L
    while (i < ByteBufferPerf.DATA_SIZE) {
      sum += data.get
      i += 1
    }
    println("sum: " + sum)
  }

  override def setUp = data.rewind()
}


object RamDiskIntBufferPerf extends Benchmark {

  val channel = new java.io.RandomAccessFile("/Volumes/ramdisk/test.dat", "r").getChannel()
  val data1: java.nio.MappedByteBuffer = channel.map(
    java.nio.channels.FileChannel.MapMode.READ_ONLY,
    0,
    math.min(channel.size(), java.lang.Integer.MAX_VALUE))

  data1.order(java.nio.ByteOrder.nativeOrder)
  val data = data1.asIntBuffer()

  override def run = {
    var i = 0
    var sum = 0L
    while (i < ByteBufferPerf.DATA_SIZE) {
      sum += data.get
      i += 1
    }
    println("sum: " + sum)
  }

  override def setUp = data.rewind()
}


object DirectUnsafePerf extends Benchmark {

  val data = DirectByteBufferPerf.data
  val unsafe = UnsafeByteBufferReader.unsafe
  val address: Long = UnsafeByteBufferReader.findAddress(data)

  override def run = {
    var offset = 0
    var sum = 0L
    while (offset < ByteBufferPerf.DATA_SIZE * 4) {
      sum += unsafe.getInt(address + offset)
      offset += 4
    }
    println("sum: " + sum)
  }
}


object RamDiskUnsafePerf extends Benchmark {

  val channel = new java.io.RandomAccessFile("/Volumes/ramdisk/test.dat", "r").getChannel()
  val data: java.nio.MappedByteBuffer = channel.map(
    java.nio.channels.FileChannel.MapMode.READ_ONLY,
    0,
    math.min(channel.size(), java.lang.Integer.MAX_VALUE))

  val unsafe = UnsafeByteBufferReader.unsafe
  val address: Long = UnsafeByteBufferReader.findAddress(data)

  override def run = {
    var offset = 0
    var sum = 0L
    while (offset < ByteBufferPerf.DATA_SIZE * 4) {
      sum += unsafe.getInt(address + offset)
      offset += 4
    }
    println("sum: " + sum)
  }
}



object UnsafeByteBufferReaderPerf extends Benchmark {

  val data = RamDiskUnsafePerf.data
  var reader: ByteBufferReader = null

  override def run = {
    var i = 0
    var sum = 0L
    while (i < ByteBufferPerf.DATA_SIZE) {
      sum += reader.getInt()
      i += 1
    }
    println("sum: " + sum)
  }

  override def setUp = { reader = new UnsafeByteBufferReader(data) }
}


object UnsafeHeapByteBufferReaderPerf extends Benchmark {

  val data = ByteBufferPerf.data
  var reader: ByteBufferReader = null

  override def run = {
    var i = 0
    var sum = 0L
    while (i < ByteBufferPerf.DATA_SIZE) {
      sum += reader.getInt()
      i += 1
    }
    println("sum: " + sum)
  }

  override def setUp = { reader = new UnsafeHeapByteBufferReader(data) }
}


/** Some helper methods to get the address and unsafe object. */
object UnsafeByteBufferReader {

  val unsafe: sun.misc.Unsafe = {
    val unsafeField = classOf[sun.misc.Unsafe].getDeclaredField("theUnsafe")
    unsafeField.setAccessible(true)
    unsafeField.get(null).asInstanceOf[sun.misc.Unsafe]
  }

  def findAddress(buffer: java.nio.ByteBuffer): Long = {
    if (buffer.hasArray) {
      val baseOffset: Long = unsafe.arrayBaseOffset(classOf[Array[Byte]])

      unsafe.addressSize() match {
        case 4 => unsafe.getInt(buffer.array(), baseOffset)
        case 8 => unsafe.getLong(buffer.array(), baseOffset)
        case _ => throw new Error("unsupported address size: " + unsafe.addressSize())
      }
    } else {
      val addressField = classOf[java.nio.Buffer].getDeclaredField("address")
      addressField.setAccessible(true)
      addressField.get(buffer).asInstanceOf[Long]
    }
  }
}



trait ByteBufferReader {
  def getInt(): Int
}


/**
 * An implementation of the ByteBufferReader using sun.misc.Unsafe. This provides very high
 * throughput read of various primitive types from a ByteBuffer, but can potentially
 * crash the JVM if the implementation is faulty.
 */
class UnsafeByteBufferReader(buf: java.nio.ByteBuffer) extends ByteBufferReader {

  private var _offset: Long = UnsafeByteBufferReader.findAddress(buf)

  override def getInt(): Int = {
    val v = UnsafeByteBufferReader.unsafe.getInt(_offset)
    _offset += 4
    v
  }
}


class UnsafeHeapByteBufferReader(buf: java.nio.ByteBuffer) extends ByteBufferReader {
  private var _arr = buf.array()
  private var _offset: Long = 16
  override def getInt(): Int = {
    val v = UnsafeByteBufferReader.unsafe.getInt(_arr, _offset)
    _offset += 4
    v
  }
}


/** Reading from int array as a baseline. */
object IntArrayPerf extends Benchmark {

  val data = new Array[Int](ByteBufferPerf.DATA_SIZE)

  override def run = {
    var i = 0
    var sum = 0L
    while (i < ByteBufferPerf.DATA_SIZE) {
      sum += data(i)
      i += 1
    }
    println("sum: " + sum)
  }
}


/** Write to a file in ramdisk */
object CreateRamDiskTestFile {
  def main(args: Array[String]) {
    val f = new java.io.FileOutputStream(new java.io.File("/Volumes/ramdisk/test.dat"))
    f.write(ByteBufferPerf.byteArray)
  }
}

ByteBufferPerf.txt

Summary:

ByteBuffer: ~ 480
ByteBuffer (direct): ~ 320
ByteBuffer (ramdisk): ~ 295

IntBuffer: ~ 330
IntBuffer (direct): ~ 85
IntBuffer (ramdisk): ~ 85

Unsafe (direct): ~ 85
Unsafe (ramdisk): ~ 77
Unsafe (using UnsafeByteBufferReaderPerf): ~ 50