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 <mme...@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 <mme...@apache.org> Author: Sijie Guo <guosi...@gmail.com> Reviewers: Enrico Olivelli <eolive...@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) {