Potential performance improvements helping Akka Http

akka-actor/src/main/scala/akka/io/SelectionHandler.scala

@@ -106,13 +106,13 @@ private[io] object SelectionHandler {
 
   private class ChannelRegistryImpl(executionContext: ExecutionContext, log: LoggingAdapter) extends ChannelRegistry {
     private[this] val selector = SelectorProvider.provider.openSelector
-    private[this] val wakeUp = new AtomicBoolean(false)
+    //private[this] val wakeUp = new AtomicBoolean(false)
 
     final val OP_READ_AND_WRITE = OP_READ | OP_WRITE // compile-time constant
 
     private[this] val select = new Task {
       def tryRun(): Unit = {
-        if (selector.select() > 0) { // This assumes select return value == selectedKeys.size
+        if (selector.select(1) > 0) { // This assumes select return value == selectedKeys.size
           val keys = selector.selectedKeys
           val iterator = keys.iterator()
           while (iterator.hasNext) {
@@ -140,7 +140,7 @@ private[io] object SelectionHandler {
           }
           keys.clear() // we need to remove the selected keys from the set, otherwise they remain selected
         }
-        wakeUp.set(false)
+        //wakeUp.set(false)
       }
 
       override def run(): Unit =
@@ -208,8 +208,8 @@ private[io] object SelectionHandler {
 
     private def execute(task: Task): Unit = {
       executionContext.execute(task)
-      if (wakeUp.compareAndSet(false, true)) // if possible avoid syscall and trade off with LOCK CMPXCHG
-        selector.wakeup()
+      //if (wakeUp.compareAndSet(false, true)) // if possible avoid syscall and trade off with LOCK CMPXCHG
+      //selector.wakeup()
     }
 
     // FIXME: Add possibility to signal failure of task to someone

akka-actor/src/main/scala/akka/io/TcpConnection.scala

@@ -373,18 +373,22 @@ private[io] abstract class TcpConnection(val tcp: TcpExt, val channel: SocketCha
     create(write)
   }
 
-  def PendingBufferWrite(commander: ActorRef, data: ByteString, ack: Event, tail: WriteCommand): PendingBufferWrite = {
-    val buffer = bufferPool.acquire()
-    try {
-      val copied = data.copyToBuffer(buffer)
-      buffer.flip()
-      new PendingBufferWrite(commander, data.drop(copied), ack, buffer, tail)
-    } catch {
-      case NonFatal(e) ⇒
-        bufferPool.release(buffer)
-        throw e
+  def PendingBufferWrite(commander: ActorRef, data: ByteString, ack: Event, tail: WriteCommand): PendingBufferWrite =
+    data match {
+      case b: ByteString.ByteString1CByteBuffer ⇒
+        new PendingBufferWrite(commander, ByteString.empty, ack, data.asByteBuffer, tail)
+      case _ ⇒
+        val buffer = bufferPool.acquire()
+        try {
+          val copied = data.copyToBuffer(buffer)
+          buffer.flip()
+          new PendingBufferWrite(commander, data.drop(copied), ack, buffer, tail)
+        } catch {
+          case NonFatal(e) ⇒
+            bufferPool.release(buffer)
+            throw e
+        }
     }
-  }
 
   class PendingBufferWrite(
     val commander: ActorRef,

akka-actor/src/main/scala/akka/util/ByteString.scala

@@ -12,7 +12,7 @@ import scala.annotation.{ tailrec, varargs }
 import scala.collection.IndexedSeqOptimized
 import scala.collection.mutable.{ Builder, WrappedArray }
 import scala.collection.immutable
-import scala.collection.immutable.{ IndexedSeq, VectorBuilder, VectorIterator }
+import scala.collection.immutable.{ IndexedSeq, Iterable, VectorBuilder, VectorIterator }
 import scala.collection.generic.CanBuildFrom
 import scala.reflect.ClassTag
 import java.nio.charset.{ Charset, StandardCharsets }
@@ -116,6 +116,24 @@ object ByteString {
     }
   }
 
+  final class ByteString1CByteBuffer(byteBuffer: ByteBuffer) extends CompactByteString {
+    def length: Int = byteBuffer.limit()
+    def apply(idx: Int): Byte = byteBuffer.get(idx)
+
+    private[akka] def byteStringCompanion: Companion = convertToRegular.byteStringCompanion
+    private[akka] def writeToOutputStream(os: ObjectOutputStream): Unit = convertToRegular.writeToOutputStream(os)
+
+    def ++(that: ByteString): ByteString = convertToRegular ++ that
+
+    def asByteBuffer: ByteBuffer = byteBuffer
+    def asByteBuffers: Iterable[ByteBuffer] = byteBuffer :: Nil
+
+    def convertToRegular: ByteString = ByteString.fromByteBuffer(byteBuffer)
+
+    def decodeString(charset: String): String = convertToRegular.decodeString(charset)
+    def decodeString(charset: Charset): String = convertToRegular.decodeString(charset)
+  }
+
   /**
    * A compact (unsliced) and unfragmented ByteString, implementation of ByteString1C.
    */

akka-stream/src/main/scala/akka/stream/impl/fusing/ActorGraphInterpreter.scala

@@ -480,14 +480,25 @@ import scala.util.control.NonFatal
 
   private var enqueueToShortCircuit: (Any) ⇒ Unit = _
 
-  lazy val interpreter: GraphInterpreter = new GraphInterpreter(mat, log, logics, connections,
-    (logic, event, handler) ⇒ {
-      val asyncInput = AsyncInput(this, logic, event, handler)
-      val currentInterpreter = GraphInterpreter.currentInterpreterOrNull
-      if (currentInterpreter == null || (currentInterpreter.context ne self))
-        self ! asyncInput
-      else enqueueToShortCircuit(asyncInput)
-    }, settings.fuzzingMode, self)
+  lazy val interpreter: GraphInterpreter =
+    if (connections.size == 31)
+      new GraphInterpreter31Impl(mat, log, logics, connections,
+        (logic, event, handler) ⇒ {
+          val asyncInput = AsyncInput(this, logic, event, handler)
+          val currentInterpreter = GraphInterpreter.currentInterpreterOrNull
+          if (currentInterpreter == null || (currentInterpreter.context ne self))
+            self ! asyncInput
+          else enqueueToShortCircuit(asyncInput)
+        }, settings.fuzzingMode, self)
+    else
+      new GraphInterpreterImpl(mat, log, logics, connections,
+        (logic, event, handler) ⇒ {
+          val asyncInput = AsyncInput(this, logic, event, handler)
+          val currentInterpreter = GraphInterpreter.currentInterpreterOrNull
+          if (currentInterpreter == null || (currentInterpreter.context ne self))
+            self ! asyncInput
+          else enqueueToShortCircuit(asyncInput)
+        }, settings.fuzzingMode, self)
 
   // TODO: really needed?
   private var subscribesPending = 0