You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2013/07/09 01:14:58 UTC
[33/43] git commit: KAFKA-872;
Socket server should set send/recv socket buffer sizes; reviewed by Jun Rao.
KAFKA-872; Socket server should set send/recv socket buffer sizes; reviewed by Jun Rao.
Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/ede85875
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/ede85875
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/ede85875
Branch: refs/heads/trunk
Commit: ede85875a3aea5723baf82736cc71bb1a04a6102
Parents: 1976ce8
Author: Joel Koshy <jj...@gmail.com>
Authored: Tue Apr 23 10:40:47 2013 -0700
Committer: Joel Koshy <jj...@gmail.com>
Committed: Tue Apr 23 10:40:47 2013 -0700
----------------------------------------------------------------------
.../main/scala/kafka/network/SocketServer.scala | 20 ++++++++++++++++----
.../main/scala/kafka/server/KafkaServer.scala | 2 ++
.../unit/kafka/network/SocketServerTest.scala | 2 ++
3 files changed, 20 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kafka/blob/ede85875/core/src/main/scala/kafka/network/SocketServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala
index 648d936..5a44c28 100644
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -37,6 +37,8 @@ class SocketServer(val brokerId: Int,
val port: Int,
val numProcessorThreads: Int,
val maxQueuedRequests: Int,
+ val sendBufferSize: Int,
+ val recvBufferSize: Int,
val maxRequestSize: Int = Int.MaxValue) extends Logging {
this.logIdent = "[Socket Server on Broker " + brokerId + "], "
private val time = SystemTime
@@ -56,7 +58,7 @@ class SocketServer(val brokerId: Int,
requestChannel.addResponseListener((id:Int) => processors(id).wakeup())
// start accepting connections
- this.acceptor = new Acceptor(host, port, processors)
+ this.acceptor = new Acceptor(host, port, processors, sendBufferSize, recvBufferSize)
Utils.newThread("kafka-acceptor", acceptor, false).start()
acceptor.awaitStartup
info("started")
@@ -128,7 +130,8 @@ private[kafka] abstract class AbstractServerThread extends Runnable with Logging
/**
* Thread that accepts and configures new connections. There is only need for one of these
*/
-private[kafka] class Acceptor(val host: String, val port: Int, private val processors: Array[Processor]) extends AbstractServerThread {
+private[kafka] class Acceptor(val host: String, val port: Int, private val processors: Array[Processor],
+ val sendBufferSize: Int, val recvBufferSize: Int) extends AbstractServerThread {
val serverChannel = openServerSocket(host, port)
/**
@@ -192,10 +195,19 @@ private[kafka] class Acceptor(val host: String, val port: Int, private val proce
* Accept a new connection
*/
def accept(key: SelectionKey, processor: Processor) {
- val socketChannel = key.channel().asInstanceOf[ServerSocketChannel].accept()
- debug("Accepted connection from " + socketChannel.socket.getInetAddress() + " on " + socketChannel.socket.getLocalSocketAddress)
+ val serverSocketChannel = key.channel().asInstanceOf[ServerSocketChannel]
+ serverSocketChannel.socket().setReceiveBufferSize(recvBufferSize)
+
+ val socketChannel = serverSocketChannel.accept()
socketChannel.configureBlocking(false)
socketChannel.socket().setTcpNoDelay(true)
+ socketChannel.socket().setSendBufferSize(sendBufferSize)
+
+ debug("Accepted connection from %s on %s. sendBufferSize [actual|requested]: [%d|%d] recvBufferSize [actual|requested]: [%d|%d]"
+ .format(socketChannel.socket.getInetAddress, socketChannel.socket.getLocalSocketAddress,
+ socketChannel.socket.getSendBufferSize, sendBufferSize,
+ socketChannel.socket.getReceiveBufferSize, recvBufferSize))
+
processor.accept(socketChannel)
}
http://git-wip-us.apache.org/repos/asf/kafka/blob/ede85875/core/src/main/scala/kafka/server/KafkaServer.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index 7298ccb..b4a57c6 100644
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -66,6 +66,8 @@ class KafkaServer(val config: KafkaConfig, time: Time = SystemTime) extends Logg
config.port,
config.numNetworkThreads,
config.queuedMaxRequests,
+ config.socketSendBufferBytes,
+ config.socketReceiveBufferBytes,
config.socketRequestMaxBytes)
socketServer.startup
http://git-wip-us.apache.org/repos/asf/kafka/blob/ede85875/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
index b347e66..94b5a2a 100644
--- a/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
+++ b/core/src/test/scala/unit/kafka/network/SocketServerTest.scala
@@ -38,6 +38,8 @@ class SocketServerTest extends JUnitSuite {
port = kafka.utils.TestUtils.choosePort,
numProcessorThreads = 1,
maxQueuedRequests = 50,
+ sendBufferSize = 300000,
+ recvBufferSize = 300000,
maxRequestSize = 50)
server.startup()