diff --git a/akka-remote/src/main/scala/akka/remote/RemoteShared.scala b/akka-remote/src/main/scala/akka/remote/RemoteShared.scala index ee4e5cf809..9fa9d1b5c0 100644 --- a/akka-remote/src/main/scala/akka/remote/RemoteShared.scala +++ b/akka-remote/src/main/scala/akka/remote/RemoteShared.scala @@ -44,4 +44,24 @@ object RemoteServerSettings { } val BACKLOG = config.getInt("akka.remote.server.backlog", 4096) + + val EXECUTION_POOL_KEEPALIVE = Duration(config.getInt("akka.remote.server.execution-pool-keepalive", 60), TIME_UNIT) + + val EXECUTION_POOL_SIZE = { + val sz = config.getInt("akka.remote.server.execution-pool-size",16) + if (sz < 1) throw new IllegalArgumentException("akka.remote.server.execution-pool-size is less than 1") + sz + } + + val MAX_CHANNEL_MEMORY_SIZE = { + val sz = config.getInt("akka.remote.server.max-channel-memory-size", 0) + if (sz < 0) throw new IllegalArgumentException("akka.remote.server.max-channel-memory-size is less than 0") + sz + } + + val MAX_TOTAL_MEMORY_SIZE = { + val sz = config.getInt("akka.remote.server.max-total-memory-size", 0) + if (sz < 0) throw new IllegalArgumentException("akka.remote.server.max-total-memory-size is less than 0") + sz + } } diff --git a/akka-remote/src/main/scala/akka/remote/netty/NettyRemoteSupport.scala b/akka-remote/src/main/scala/akka/remote/netty/NettyRemoteSupport.scala index 893b22b059..3fab1b20c1 100644 --- a/akka-remote/src/main/scala/akka/remote/netty/NettyRemoteSupport.scala +++ b/akka-remote/src/main/scala/akka/remote/netty/NettyRemoteSupport.scala @@ -33,6 +33,7 @@ import org.jboss.netty.handler.codec.frame.{ LengthFieldBasedFrameDecoder, Lengt import org.jboss.netty.handler.codec.compression.{ ZlibDecoder, ZlibEncoder } import org.jboss.netty.handler.codec.protobuf.{ ProtobufDecoder, ProtobufEncoder } import org.jboss.netty.handler.timeout.{ ReadTimeoutHandler, ReadTimeoutException } +import org.jboss.netty.handler.execution.{ OrderedMemoryAwareThreadPoolExecutor, ExecutionHandler } import org.jboss.netty.util.{ TimerTask, Timeout, HashedWheelTimer } import org.jboss.netty.handler.ssl.SslHandler @@ -753,9 +754,17 @@ class RemoteServerPipelineFactory( case "zlib" => (new ZlibEncoder(ZLIB_COMPRESSION_LEVEL) :: Nil, new ZlibDecoder :: Nil) case _ => (Nil, Nil) } - + val execution = new ExecutionHandler( + new OrderedMemoryAwareThreadPoolExecutor( + EXECUTION_POOL_SIZE, + MAX_CHANNEL_MEMORY_SIZE, + MAX_TOTAL_MEMORY_SIZE, + EXECUTION_POOL_KEEPALIVE.length, + EXECUTION_POOL_KEEPALIVE.unit + ) + ) val remoteServer = new RemoteServerHandler(name, openChannels, loader, server) - val stages: List[ChannelHandler] = dec ::: lenDec :: protobufDec :: enc ::: lenPrep :: protobufEnc :: remoteServer :: Nil + val stages: List[ChannelHandler] = dec ::: lenDec :: protobufDec :: enc ::: lenPrep :: protobufEnc :: execution :: remoteServer :: Nil new StaticChannelPipeline(stages: _*) } } @@ -856,8 +865,6 @@ class RemoteServerHandler( } private def handleRemoteMessageProtocol(request: RemoteMessageProtocol, channel: Channel) = { - //FIXME we should definitely spawn off this in a thread pool or something, - // potentially using Actor.spawn or something similar request.getActorInfo.getActorType match { case SCALA_ACTOR => dispatchToActor(request, channel) case TYPED_ACTOR => dispatchToTypedActor(request, channel) diff --git a/akka-remote/src/test/scala/config/ConfigSpec.scala b/akka-remote/src/test/scala/config/ConfigSpec.scala index 7ba5193f6f..e37edcfc34 100644 --- a/akka-remote/src/test/scala/config/ConfigSpec.scala +++ b/akka-remote/src/test/scala/config/ConfigSpec.scala @@ -36,6 +36,10 @@ class ConfigSpec extends WordSpec with MustMatchers { getBool("akka.remote.ssl.debug") must equal(None) getBool("akka.remote.ssl.service") must equal(None) getInt("akka.remote.zlib-compression-level") must equal(Some(6)) + getInt("akka.remote.server.execution-pool-size") must equal(Some(16)) + getInt("akka.remote.server.execution-pool-keepalive") must equal(Some(60)) + getInt("akka.remote.server.max-channel-memory-size") must equal(Some(0)) + getInt("akka.remote.server.max-total-memory-size") must equal(Some(0)) } } } diff --git a/config/akka-reference.conf b/config/akka-reference.conf index 458f937e5e..b27c383de6 100644 --- a/config/akka-reference.conf +++ b/config/akka-reference.conf @@ -142,6 +142,10 @@ akka { require-cookie = off # Should the remote server require that it peers share the same secure-cookie (defined in the 'remote' section)? untrusted-mode = off # Enable untrusted mode for full security of server managed actors, allows untrusted clients to connect. backlog = 4096 # Sets the size of the connection backlog + execution-pool-keepalive = 60# Length in akka.time-unit how long core threads will be kept alive if idling + execution-pool-size = 16# Size of the core pool of the remote execution unit + max-channel-memory-size = 0 # Maximum channel size, 0 for off + max-total-memory-size = 0 # Maximum total size of all channels, 0 for off } client {