kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jjko...@apache.org
Subject git commit: KAFKA-872; Socket server should set send/recv socket buffer sizes; reviewed by Jun Rao.
Date Tue, 23 Apr 2013 17:40:59 GMT
Updated Branches:
  refs/heads/0.8 1976ce806 -> ede85875a


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/0.8
Commit: ede85875a3aea5723baf82736cc71bb1a04a6102
Parents: 1976ce8
Author: Joel Koshy <jjkoshy@gmail.com>
Authored: Tue Apr 23 10:40:47 2013 -0700
Committer: Joel Koshy <jjkoshy@gmail.com>
Committed: Tue Apr 23 10:40:47 2013 -0700

----------------------------------------------------------------------
 .../main/scala/kafka/network/SocketServer.scala    |   20 ++++++++++++---
 core/src/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()
 


Mime
View raw message