You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by si...@apache.org on 2018/08/21 22:24:05 UTC
[bookkeeper] branch branch-4.8 updated: Allow to configure num of
Netty IO threads in client and bookie
This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch branch-4.8
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/branch-4.8 by this push:
new ffdd631 Allow to configure num of Netty IO threads in client and bookie
ffdd631 is described below
commit ffdd63197b69f182a55ec1275af38df36e9bab7a
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Tue Aug 21 15:12:21 2018 -0700
Allow to configure num of Netty IO threads in client and bookie
Currently the number of IO threads for client and bookie are set to `2 * Runtime.getRuntime().availableProcessors()`. Added configuration options to tune it.
Author: Matteo Merli <mm...@apache.org>
Author: Sijie Guo <gu...@gmail.com>
Reviewers: Enrico Olivelli <eo...@gmail.com>, Sijie Guo <si...@apache.org>, Venkateswararao Jujjuri (JV) <None>
This closes #1612 from merlimat/configure-io-threads
(cherry picked from commit b9916d3d87208af36be6a7b061a594a33c897271)
Signed-off-by: Sijie Guo <si...@apache.org>
---
.../org/apache/bookkeeper/client/BookKeeper.java | 8 +++--
.../bookkeeper/conf/ClientConfiguration.java | 34 +++++++++++++++++++++-
.../bookkeeper/conf/ServerConfiguration.java | 27 +++++++++++++++++
.../apache/bookkeeper/proto/BookieNettyServer.java | 2 +-
4 files changed, 66 insertions(+), 5 deletions(-)
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
index d187b41..024295f 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/client/BookKeeper.java
@@ -465,7 +465,7 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
// initialize event loop group
if (null == eventLoopGroup) {
- this.eventLoopGroup = getDefaultEventLoopGroup();
+ this.eventLoopGroup = getDefaultEventLoopGroup(conf);
this.ownEventLoopGroup = true;
} else {
this.eventLoopGroup = eventLoopGroup;
@@ -1392,6 +1392,7 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
*/
public void asyncIsClosed(long lId, final IsClosedCallback cb, final Object ctx){
ledgerManager.readLedgerMetadata(lId, new GenericCallback<LedgerMetadata>(){
+ @Override
public void operationComplete(int rc, LedgerMetadata lm){
if (rc == BKException.Code.OK) {
cb.isClosedComplete(rc, lm.isClosed(), ctx);
@@ -1421,6 +1422,7 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
final Result result = new Result();
final IsClosedCallback cb = new IsClosedCallback(){
+ @Override
public void isClosedComplete(int rc, boolean isClosed, Object ctx){
result.isClosed = isClosed;
result.rc = rc;
@@ -1564,9 +1566,9 @@ public class BookKeeper implements org.apache.bookkeeper.client.api.BookKeeper {
Counter getAddOpUrCounter() {
return addOpUrCounter;
}
- static EventLoopGroup getDefaultEventLoopGroup() {
+ static EventLoopGroup getDefaultEventLoopGroup(ClientConfiguration conf) {
ThreadFactory threadFactory = new DefaultThreadFactory("bookkeeper-io");
- final int numThreads = Runtime.getRuntime().availableProcessors() * 2;
+ final int numThreads = conf.getNumIOThreads();
if (SystemUtils.IS_OS_LINUX) {
try {
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java
index 340b40b..c902db8 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java
@@ -143,8 +143,9 @@ public class ClientConfiguration extends AbstractConfiguration<ClientConfigurati
protected static final String GET_BOOKIE_INFO_TIMEOUT_SECS = "getBookieInfoTimeoutSecs";
protected static final String START_TLS_TIMEOUT_SECS = "startTLSTimeoutSecs";
- // Number Woker Threads
+ // Number of Threads
protected static final String NUM_WORKER_THREADS = "numWorkerThreads";
+ protected static final String NUM_IO_THREADS = "numIOThreads";
// Ensemble Placement Policy
protected static final String ENSEMBLE_PLACEMENT_POLICY = "ensemblePlacementPolicy";
@@ -851,6 +852,37 @@ public class ClientConfiguration extends AbstractConfiguration<ClientConfigurati
}
/**
+ * Get the number of IO threads. This is the number of
+ * threads used by Netty to handle TCP connections.
+ *
+ * @return the number of IO threads
+ */
+ public int getNumIOThreads() {
+ return getInt(NUM_IO_THREADS, 2 * Runtime.getRuntime().availableProcessors());
+ }
+
+ /**
+ * Set the number of IO threads.
+ *
+ * <p>
+ * This is the number of threads used by Netty to handle TCP connections.
+ * </p>
+ *
+ * <p>
+ * NOTE: setting the number of IO threads after BookKeeper object is constructed
+ * will not take any effect on the number of threads in the pool.
+ * </p>
+ *
+ * @see #getNumIOThreads()
+ * @param numThreads number of IO threads used for bookkeeper
+ * @return client configuration
+ */
+ public ClientConfiguration setNumIOThreads(int numThreads) {
+ setProperty(NUM_IO_THREADS, numThreads);
+ return this;
+ }
+
+ /**
* Get the period of time after which a speculative entry read should be triggered.
* A speculative entry read is sent to the next replica bookie before
* an error or response has been received for the previous entry read request.
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java
index 7e14f2e..db40e0f 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ServerConfiguration.java
@@ -106,6 +106,7 @@ public class ServerConfiguration extends AbstractConfiguration<ServerConfigurati
protected static final String SERVER_SOCK_LINGER = "serverTcpLinger";
protected static final String SERVER_WRITEBUFFER_LOW_WATER_MARK = "serverWriteBufferLowWaterMark";
protected static final String SERVER_WRITEBUFFER_HIGH_WATER_MARK = "serverWriteBufferHighWaterMark";
+ protected static final String SERVER_NUM_IO_THREADS = "serverNumIOThreads";
// Zookeeper Parameters
protected static final String ZK_RETRY_BACKOFF_START_MS = "zkRetryBackoffStartMs";
@@ -1116,6 +1117,32 @@ public class ServerConfiguration extends AbstractConfiguration<ServerConfigurati
}
/**
+ * Get the number of IO threads. This is the number of
+ * threads used by Netty to handle TCP connections.
+ *
+ * @return the number of IO threads
+ */
+ public int getServerNumIOThreads() {
+ return getInt(SERVER_NUM_IO_THREADS, 2 * Runtime.getRuntime().availableProcessors());
+ }
+
+ /**
+ * Set the number of IO threads.
+ *
+ * <p>
+ * This is the number of threads used by Netty to handle TCP connections.
+ * </p>
+ *
+ * @see #getNumIOThreads()
+ * @param numThreads number of IO threads used for bookkeeper
+ * @return client configuration
+ */
+ public ServerConfiguration setServerNumIOThreads(int numThreads) {
+ setProperty(SERVER_NUM_IO_THREADS, Integer.toString(numThreads));
+ return this;
+ }
+
+ /**
* Timeout to drain the socket on close.
*
* @return socket linger setting
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java
index d687a5c..1cbb345 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java
@@ -112,7 +112,7 @@ class BookieNettyServer {
if (!conf.isDisableServerSocketBind()) {
ThreadFactory threadFactory = new DefaultThreadFactory("bookie-io");
- final int numThreads = Runtime.getRuntime().availableProcessors() * 2;
+ final int numThreads = conf.getServerNumIOThreads();
EventLoopGroup eventLoopGroup;
if (SystemUtils.IS_OS_LINUX) {