[
https://issues.apache.org/jira/browse/HDFS-17909?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18081555#comment-18081555
]
ASF GitHub Bot commented on HDFS-17909:
---------------------------------------
ZanderXu commented on code in PR #8448:
URL: https://github.com/apache/hadoop/pull/8448#discussion_r3256057730
##########
hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java:
##########
@@ -540,21 +569,17 @@ public void initAsyncThreadPools(Configuration
configuration) {
asyncResponderCount, new AsyncThreadFactory("Router Async Responder
#"));
}
AsyncRpcProtocolPBUtil.setAsyncResponderExecutor(routerAsyncResponderExecutor);
-
- if (routerDefaultAsyncHandlerExecutor == null) {
- LOG.info("init router async default executor handler count: {}",
asyncHandlerCountDefault);
- routerDefaultAsyncHandlerExecutor = Executors.newFixedThreadPool(
- asyncHandlerCountDefault, new AsyncThreadFactory("Router Async
Default Handler #"));
- }
}
- private void initNsAsyncHandlerCount() {
- String configNsHandler =
conf.get(DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY,
+ private Map<String, Integer> parseNsAsyncHandlerCount(Configuration config) {
+ String configNsHandler =
config.get(DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_KEY,
DFS_ROUTER_ASYNC_RPC_NS_HANDLER_COUNT_DEFAULT);
+ Map<String, Integer> nsAsyncHandlerCount = new HashMap<>();
if (StringUtils.isEmpty(configNsHandler)) {
- LOG.error(
- "The value of config key: {} is empty. Will use default conf.",
+ LOG.debug("No per-namespace async handler counts configured ({}). "
Review Comment:
change the `debug` to `info` or `warn`.
##########
hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java:
##########
@@ -504,34 +508,59 @@ public RouterRpcServer(Configuration conf, Router router,
* @param configuration the configuration.
*/
public void initAsyncThreadPools(Configuration configuration) {
- LOG.info("Begin initialize asynchronous handler and responder thread
pool.");
- initNsAsyncHandlerCount();
Set<String> allConfiguredNS =
FederationUtil.getAllConfiguredNS(configuration);
- Set<String> unassignedNS = new HashSet<>();
allConfiguredNS.add(CONCURRENT_NS);
-
+ Map<String, Integer> nsAsyncActiveHandlerCount =
parseNsAsyncHandlerCount(configuration);
+ initAsyncHandlerThreadPools(configuration, allConfiguredNS,
nsAsyncActiveHandlerCount);
+ initAsyncResponderThreadPools(configuration);
+ }
+
+ private void initAsyncHandlerThreadPools(Configuration configuration,
+ Set<String> allConfiguredNS, Map<String, Integer> nsAsyncHandlerCount) {
+ LOG.info("Initializing asynchronous handler thread pools");
+ int asyncQueueSize = configuration.getInt(DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE,
+ DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE_DEFAULT);
+ if (asyncQueueSize < 1) {
+ throw new IllegalArgumentException("Async queue size must be at least
1");
+ }
+ int asyncHandlerCountDefault =
configuration.getInt(DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY,
+ DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_DEFAULT);
for (String nsId : allConfiguredNS) {
int dedicatedHandlers = nsAsyncHandlerCount.getOrDefault(nsId, 0);
- LOG.info("Dedicated handlers {} for ns {} ", dedicatedHandlers, nsId);
- if (dedicatedHandlers > 0) {
- initAsyncHandlerThreadPools4Ns(nsId, dedicatedHandlers);
- LOG.info("Assigned {} async handlers to nsId {} ", dedicatedHandlers,
nsId);
+ if (dedicatedHandlers <= 0) {
+ dedicatedHandlers = asyncHandlerCountDefault;
+ LOG.info("Use default async handler count {} for ns {} to init
Executors.",
+ asyncHandlerCountDefault, nsId);
} else {
- unassignedNS.add(nsId);
+ LOG.info("Dedicated handlers {} for ns {} to init Executors",
dedicatedHandlers, nsId);
}
- }
-
- int asyncHandlerCountDefault =
configuration.getInt(DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY,
- DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_DEFAULT);
- if (!unassignedNS.isEmpty()) {
- LOG.warn("Async handler unassigned ns: {}", unassignedNS);
- LOG.info("Use default async handler count {} for unassigned ns.",
asyncHandlerCountDefault);
- for (String nsId : unassignedNS) {
- initAsyncHandlerThreadPools4Ns(nsId, asyncHandlerCountDefault);
+ if (dedicatedHandlers > 0) {
Review Comment:
this check is useless.
##########
hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcServer.java:
##########
@@ -504,34 +508,59 @@ public RouterRpcServer(Configuration conf, Router router,
* @param configuration the configuration.
*/
public void initAsyncThreadPools(Configuration configuration) {
- LOG.info("Begin initialize asynchronous handler and responder thread
pool.");
- initNsAsyncHandlerCount();
Set<String> allConfiguredNS =
FederationUtil.getAllConfiguredNS(configuration);
- Set<String> unassignedNS = new HashSet<>();
allConfiguredNS.add(CONCURRENT_NS);
-
+ Map<String, Integer> nsAsyncActiveHandlerCount =
parseNsAsyncHandlerCount(configuration);
+ initAsyncHandlerThreadPools(configuration, allConfiguredNS,
nsAsyncActiveHandlerCount);
+ initAsyncResponderThreadPools(configuration);
+ }
+
+ private void initAsyncHandlerThreadPools(Configuration configuration,
+ Set<String> allConfiguredNS, Map<String, Integer> nsAsyncHandlerCount) {
+ LOG.info("Initializing asynchronous handler thread pools");
+ int asyncQueueSize = configuration.getInt(DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE,
+ DFS_ROUTER_ASYNC_RPC_QUEUE_SIZE_DEFAULT);
+ if (asyncQueueSize < 1) {
+ throw new IllegalArgumentException("Async queue size must be at least
1");
+ }
+ int asyncHandlerCountDefault =
configuration.getInt(DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_KEY,
+ DFS_ROUTER_ASYNC_RPC_HANDLER_COUNT_DEFAULT);
Review Comment:
throw IllegalArgumentException if `asyncHandlerCountDefault` <= 0
> [ARR] AsyncRouterHandlerExecutors should use bounded queue
> ----------------------------------------------------------
>
> Key: HDFS-17909
> URL: https://issues.apache.org/jira/browse/HDFS-17909
> Project: Hadoop HDFS
> Issue Type: Improvement
> Components: rbf
> Reporter: Felix N
> Assignee: Felix N
> Priority: Major
> Labels: pull-request-available
>
> AsyncRouterHandlerExecutors should use bounded queues to prevent queues from
> growing without a limit.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
