Merge pull request #49 from penland365/upgrade-buffer

Introduced the Buf type as Readable / Writeable Array Buffer.

.jvmopts

@@ -0,0 +1,12 @@
+-Dfile.encoding=UTF8
+-Xms1G
+-Xmx3G
+-XX:ReservedCodeCacheSize=250M
+-XX:+TieredCompilation
+-XX:-UseGCOverheadLimit
+# effectively adds GC to Perm space
+-XX:+CMSClassUnloadingEnabled
+# must be enabled for CMSClassUnloadingEnabled to work
+-XX:+UseConcMarkSweepGC
+
+-Dio.netty.leakDetection.level=advanced

benchmark/src/main/scala/roc/benchmark/PacketDecoderBenchmarks.scala

@@ -37,7 +37,6 @@ class PacketDecoderBenchmarks extends TestData {
   }
 }
 
-
 abstract class TestData {
 
   object Decoders extends PacketDecoderImplicits

benchmark/src/main/scala/roc/benchmark/buffer/BufBenchmarks.scala

@@ -0,0 +1,54 @@
+package roc
+package postgresql
+package transport
+
+import java.util.concurrent.TimeUnit
+import org.openjdk.jmh.annotations._
+
+@Fork(2)
+@State(Scope.Thread)
+class BufBenchmarks extends BufferTestData {
+
+  var buf: Buf = Buf(0)
+
+  @Setup(Level.Invocation)
+  def instantiateBuf(): Unit = {
+    buf = Buf(4)
+    ()
+  }
+
+  @TearDown(Level.Invocation)
+  def releaseBuf(): Unit = buf.release()
+
+  @Benchmark
+  @BenchmarkMode(Array(Mode.Throughput))
+  @OutputTimeUnit(TimeUnit.MICROSECONDS)
+  def measureBufAllocationAndIntWrite(): Unit = {
+    val u = buf.writeInt(TestInt)
+    u
+  }
+}
+
+@Fork(2)
+@State(Scope.Thread)
+class BufWriteCStringBenchmark extends BufferTestData {
+
+  var buf: Buf = Buf(0)
+
+  @Setup(Level.Invocation)
+  def allocateBuf(): Unit = {
+    buf = Buf(TestStringByteLength)
+    ()
+  }
+
+  @TearDown(Level.Invocation)
+  def releaseBuf(): Unit = buf.release()
+
+  @Benchmark
+  @BenchmarkMode(Array(Mode.Throughput))
+  @OutputTimeUnit(TimeUnit.MICROSECONDS)
+  def measureBufCStringWrite(): Unit = {
+    val u = buf.writeCStyleString(TestString)
+    u
+  }
+}

benchmark/src/main/scala/roc/benchmark/buffer/BufReaderBenchmarks.scala

@@ -0,0 +1,31 @@
+package roc
+package postgresql
+package transport
+
+import java.util.concurrent.TimeUnit
+import org.openjdk.jmh.annotations._
+
+@Fork(1)
+@State(Scope.Thread)
+class BufIntReaderBenchmark extends BufferTestData {
+
+  var buf = Buf(0)
+
+  @Setup(Level.Invocation)
+  def instantiateBuf(): Unit = {
+    buf = Buf(4)
+    buf.writeInt(TestInt)
+    ()
+  }
+
+  @TearDown(Level.Invocation)
+  def releaseBuf(): Unit = buf.release()
+
+  @Benchmark
+  @BenchmarkMode(Array(Mode.Throughput))
+  @OutputTimeUnit(TimeUnit.MICROSECONDS)
+  def measureBufReadInt(): Int = {
+    buf.readInt
+  }
+
+}

benchmark/src/main/scala/roc/benchmark/buffer/BufferReaderBenchmarks.scala

@@ -0,0 +1,28 @@
+package roc
+package postgresql
+package transport
+
+import java.util.concurrent.TimeUnit
+import org.openjdk.jmh.annotations._
+
+@Fork(2)
+@State(Scope.Thread)
+class BufferReaderReadIntBenchmarks extends BufferTestData {
+
+  var br = BufferReader(Array.empty[Byte])
+
+  @Setup(Level.Invocation)
+  def allocateBuffer(): Unit = {
+    val bw = BufferWriter(new Array[Byte](4))
+    bw.writeInt(TestInt)
+    br = BufferReader(bw.toBytes)
+    ()
+  }
+
+  @Benchmark
+  @BenchmarkMode(Array(Mode.Throughput))
+  @OutputTimeUnit(TimeUnit.MICROSECONDS)
+  def measureBufferReaderReadInt(): Int = {
+    br.readInt
+  }
+}

benchmark/src/main/scala/roc/benchmark/buffer/BufferTestData.scala

@@ -0,0 +1,11 @@
+package roc
+package postgresql
+package transport
+
+import java.nio.charset.StandardCharsets
+
+abstract class BufferTestData {
+  val TestInt = 71
+  val TestString = "␂䤥됳ⲟ鮩躾へ囙엊ꡓ쿟䏾뫱㳫䕣찵䥵총몫Ғ칆縏洰둈Ã糧舝裂ꚦ鯅晔뭿栛鍯楾ꑜ辔ᖊ쐡퍽䌡鏹탪ຣ剻䅧୷㒒凹ᱣ浤⯵ϵ蒤ꘞ觲᜴씻꠩ಯ힭ཷ秞囬榙ꥲ确텳꩞邀ḷ菲裁劋謆凰겕렶謱鴆◛Ⱬı熳샹칻"
+  val TestStringByteLength = TestString.getBytes(StandardCharsets.UTF_8).length + 1
+}

benchmark/src/main/scala/roc/benchmark/buffer/BufferWriterBenchmarks.scala

@@ -0,0 +1,43 @@
+package roc.postgresql.transport
+
+import java.util.concurrent.TimeUnit
+import org.openjdk.jmh.annotations._
+
+@Fork(2)
+@State(Scope.Thread)
+class BufferWriterBenchmarks extends BufferTestData {
+
+  var buf = BufferWriter(Array.empty[Byte])
+
+  @Setup(Level.Invocation)
+  def allocateBuffer(): Unit = {
+    buf = BufferWriter(new Array[Byte](4))
+    ()
+  }
+
+  @Benchmark
+  @BenchmarkMode(Array(Mode.Throughput))
+  @OutputTimeUnit(TimeUnit.MICROSECONDS)
+  def measureBufAllocationAndIntWrite(): BufferWriter = {
+    buf.writeInt(TestInt)
+  }
+}
+
+@Fork(2)
+@State(Scope.Thread)
+class BufferWriterCStringBenchmark extends BufferTestData {
+  var buf = BufferWriter(Array.empty[Byte])
+
+  @Setup(Level.Invocation)
+  def allocateBuffer(): Unit = {
+    buf = BufferWriter(new Array[Byte](TestStringByteLength))
+    ()
+  }
+
+  @Benchmark
+  @BenchmarkMode(Array(Mode.Throughput))
+  @OutputTimeUnit(TimeUnit.MICROSECONDS)
+  def measureCStyleStringWrite(): BufferWriter = {
+    buf.writeNullTerminatedString(TestString)
+  }
+}

build.sbt

@@ -58,7 +58,7 @@ lazy val catsVersion = "0.4.1"
 
 lazy val finagleVersion = "6.34.0"
 
-lazy val nettyVersion = "4.1.0.CR2"
+lazy val nettyVersion = "4.0.36.Final"
 
 lazy val roc = project.in(file("."))
   .settings(moduleName := "root")

core/src/main/scala/roc/postgresql/transport/Buf.scala

@@ -0,0 +1,189 @@
+package roc
+package postgresql
+package transport
+
+import io.netty.buffer.{ByteBuf, Unpooled}
+import java.nio.charset.{Charset, StandardCharsets}
+
+/** A Buf based on a Netty-4 ByteBuffer
+  *
+  * @note As of this moment we are using Unpooled Buffers, but this may change in the near
+  *     future.
+  */
+private[roc] object Buf {
+
+  /** Creates a buf w/ the specified length
+    * @length the capacity of the Buffer
+    * @returns a [[Buf]] with an initial capacity set to length
+    */
+  def apply(length: Int): Buf     = new Buf(Unpooled.buffer(length))
+
+  /** Creates a buf w/ the length of the bytes, and then writes the bytes to the Buffer
+    * @args an [[Array[Byte]] to write to the Buffer
+    * @returns a [[Buf]] with the bytes already written
+    */
+  def apply(xs: Array[Byte]): Buf = {
+    val buffer = Unpooled.buffer(xs.length)
+    buffer.writeBytes(xs)
+    new Buf(buffer)
+  }
+
+  /** Computes the length of an Array needed to hold all the bytes from a String + NULL
+    * @args s the String to calculate the length of
+    * @args charset the Charset to base this calculation off of ( defaults to UTF-8 )
+    * @returns length an [[Int]] representing the length of an array
+    * @note a NULL terminated string is referred to as a C-Style string
+    */
+  def lengthOfCStyleString(s: String, charset: Charset = StandardCharsets.UTF_8): Int =
+    s.getBytes(charset).length + 1
+
+  /** Helper method for lenghtOfCStyleString
+    */
+  def lengthOfCStyleStrings(xs: List[String], charset: Charset = StandardCharsets.UTF_8): Int =
+    xs match {
+      case h :: t => xs.map(x => lengthOfCStyleString(x, charset)).reduce(_ + _)
+      case t      => 0
+    }
+}
+
+/** A Buffer for reading and writing primatives to an Array
+  * @args underlying a to utilize. 
+  * @note unlike the existing Buffer, this is both read and write compatible
+  */
+private[roc] final class Buf private(private[this] val underlying: ByteBuf) {
+
+  /** Read one byte from the array
+    * @returns Byte
+    */
+  def readByte: Byte = underlying.readByte
+
+  /** Reads the number of bytes of the given as an argument from the Buffer
+    * @args count the number of bytes to read
+    * @returns bytes an [[Array[Byte]] with the requested number of Bytes
+    */
+  def readBytes(length: Int): Array[Byte] = {
+    val xs = Array.fill[Byte](length)(0x00)
+    underlying.readBytes(xs)
+    xs
+  }
+
+  /** Reads a C-Style String from the Buffer
+    * @args charset the Charset to use in decoding Bytes. Defaults to UTF-8
+    * @returns the converted String
+    */
+  def readCStyleString(charset: Charset = StandardCharsets.UTF_8): String = {
+    val idx = underlying.bytesBefore(0x00)
+    val xs  = Array.fill[Byte](idx - underlying.readerIndex)(0x00)
+    underlying.readBytes(xs)
+    new String(xs, charset)
+  }
+
+  /** Reads a Double from the Buffer
+    * @returns the double read
+    */
+  def readDouble: Double = underlying.readDouble
+
+  /** Reads a Float from the Buffer
+    * @returns the float read
+    */
+  def readFloat: Float = underlying.readFloat
+
+  /** Reads an Int from the Buffer
+    * @returns the int read
+    */
+  def readInt: Int = underlying.readInt
+
+  /** Reads a Long from the Buffer
+    * @returns the long read
+    */
+  def readLong: Long = underlying.readLong
+
+  /** Reads a Short from the Buffer
+    * @returns the short read
+    */
+  def readShort: Short = underlying.readShort
+
+  /** Convertes the readable portion of this Buffer to an [[Array[Byte]]
+    * @returns xs the viewable portion of this buffer as a byte array
+    */
+  def toBytes: Array[Byte] = {
+    val length = underlying.writerIndex - underlying.readerIndex
+    val xs = Array.fill[Byte](length)(0x00)
+    underlying.readBytes(xs)
+    xs
+  }
+
+  /** Writes a byte to the buffer
+    * @args b to the Byte to write
+    */
+  def writeByte(b: Byte): Unit = {
+    underlying.writeByte(b)
+    ()
+  }
+
+  /** Writes an Array of bytes to the buffer
+    * @args xs an [[Array[Byte]]
+    */
+  def writeBytes(xs: Array[Byte]): Unit = {
+    underlying.writeBytes(xs)
+    ()
+  }
+
+  /** Writes a String to the Buffer using the given Charset, then writes a terminated NULL
+    * @args s the String to write to the Buffer
+    * @args charset the Charset used in coverting a Char => Byte(s)
+    */
+  def writeCStyleString(s: String, charset: Charset = StandardCharsets.UTF_8): Unit = {
+    val bytes = s.getBytes(charset)
+    underlying.writeBytes(bytes)
+    underlying.writeZero(1)
+    ()
+  }
+
+  /** Writes an 8-Byte Double to Buffer
+    * @args d the double to write
+    */
+  def writeDouble(d: Double): Unit = {
+    underlying.writeDouble(d)
+    ()
+  }
+
+  /** Writes a 4-Byte Float to Buffer
+    * @args f the float to write
+    */
+  def writeFloat(f: Float): Unit = {
+    underlying.writeFloat(f)
+    ()
+  }
+
+  /** Writes a 4-Byte Int to Buffer
+    * @args i the long to write
+    */
+  def writeInt(i: Int): Unit = {
+    underlying.writeInt(i)
+    ()
+  }
+
+  /** Writes an 8-Byte long to Buffer
+    * @args l the long to write
+    */
+  def writeLong(l: Long): Unit = {
+    underlying.writeLong(l)
+    ()
+  }
+
+  /** Writes a 1-Byte NULL (0x00) to the buffer
+    */
+  def writeNull(): Unit = {
+    underlying.writeZero(1)
+    ()
+  }
+
+  /** Writes a 2-Byte short to Buffer
+    * @args s the short to write
+    */
+  def writeShort(s: Short): Unit = {
+    underlying.writeShort(s)
+    ()
+  }
+}