Skip to content

Commit 4d59b44

Browse files
hfutatzhanghbhfutatzhanghb
committed
fix review
1 parent 3065a2f commit 4d59b44

File tree

3 files changed

+38
-130
lines changed

3 files changed

+38
-130
lines changed

‎hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/fairness/RouterAsyncRpcFairnessPolicyController.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.slf4j.LoggerFactory;
2525

2626
import java.util.Set;
27+
import java.util.concurrent.TimeUnit;
2728

2829
import static org.apache.hadoop.hdfs.server.federation.fairness.RouterRpcFairnessConstants.CONCURRENT_NS;
2930
import static org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys.
@@ -70,4 +71,20 @@ public void init(Configuration conf) throws IllegalArgumentException {
7071
private static void logAssignment(String nsId, int count) {
7172
LOG.info("Assigned {} permits to nsId {} ", count, nsId);
7273
}
74+
75+
@Override
76+
public boolean acquirePermit(String nsId) {
77+
if (nsId.equals(CONCURRENT_NS)) {
78+
return true;
79+
}
80+
return super.acquirePermit(nsId);
81+
}
82+
83+
@Override
84+
public void releasePermit(String nsId) {
85+
if (nsId.equals(CONCURRENT_NS)) {
86+
return;
87+
}
88+
super.releasePermit(nsId);
89+
}
7390
}

‎hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1024,7 +1024,7 @@ public Object invokeSingle(final String nsId, RemoteMethod method)
10241024
throws IOException {
10251025
UserGroupInformation ugi = RouterRpcServer.getRemoteUser();
10261026
RouterRpcFairnessPolicyController controller = getRouterRpcFairnessPolicyController();
1027-
acquirePermit(nsId, ugi, method, controller);
1027+
acquirePermit(nsId, ugi, method.getMethodName(), controller);
10281028
try {
10291029
boolean isObserverRead = isObserverReadEligible(nsId, method.getMethod());
10301030
List<? extends FederationNamenodeContext> nns = getOrderedNamenodes(nsId, isObserverRead);
@@ -1199,7 +1199,7 @@ public <R extends RemoteLocationContext, T> RemoteResult invokeSequential(
11991199
boolean isObserverRead = isObserverReadEligible(ns, m);
12001200
List<? extends FederationNamenodeContext> namenodes =
12011201
getOrderedNamenodes(ns, isObserverRead);
1202-
acquirePermit(ns, ugi, remoteMethod, controller);
1202+
acquirePermit(ns, ugi, remoteMethod.getMethodName(), controller);
12031203
try {
12041204
Class<?> proto = remoteMethod.getProtocol();
12051205
Object[] params = remoteMethod.getParams(loc);
@@ -1579,7 +1579,7 @@ protected static <T extends RemoteLocationContext, R> Map<T, R> postProcessResul
15791579
return invokeSingle(locations.iterator().next(), method);
15801580
}
15811581
RouterRpcFairnessPolicyController controller = getRouterRpcFairnessPolicyController();
1582-
acquirePermit(CONCURRENT_NS, ugi, method, controller);
1582+
acquirePermit(CONCURRENT_NS, ugi, method.getMethodName(), controller);
15831583

15841584
List<T> orderedLocations = new ArrayList<>();
15851585
List<Callable<Object>> callables = new ArrayList<>();
@@ -1758,7 +1758,7 @@ public <T extends RemoteLocationContext, R> List<RemoteResult<T, R>> invokeSingl
17581758
final List<? extends FederationNamenodeContext> namenodes =
17591759
getOrderedNamenodes(ns, isObserverRead);
17601760
RouterRpcFairnessPolicyController controller = getRouterRpcFairnessPolicyController();
1761-
acquirePermit(ns, ugi, method, controller);
1761+
acquirePermit(ns, ugi, method.getMethodName(), controller);
17621762
try {
17631763
Class<?> proto = method.getProtocol();
17641764
Object[] paramList = method.getParams(location);
@@ -1829,12 +1829,12 @@ private String getNameserviceForBlockPoolId(final String bpId)
18291829
*
18301830
* @param nsId Identifier of the block pool.
18311831
* @param ugi UserGroupIdentifier associated with the user.
1832-
* @param m Remote method that needs to be invoked.
1832+
* @param methodName The name of remote method that needs to be invoked.
18331833
* @param controller fairness policy controller to acquire permit from
18341834
* @throws IOException If permit could not be acquired for the nsId.
18351835
*/
18361836
protected void acquirePermit(final String nsId, final UserGroupInformation ugi,
1837-
final RemoteMethod m, RouterRpcFairnessPolicyController controller)
1837+
final String methodName, RouterRpcFairnessPolicyController controller)
18381838
throws IOException {
18391839
if (controller != null) {
18401840
if (!controller.acquirePermit(nsId)) {
@@ -1845,7 +1845,7 @@ protected void acquirePermit(final String nsId, final UserGroupInformation ugi,
18451845
}
18461846
incrRejectedPermitForNs(nsId);
18471847
LOG.debug("Permit denied for ugi: {} for method: {}",
1848-
ugi, m.getMethodName());
1848+
ugi, methodName);
18491849
String msg =
18501850
"Router " + router.getRouterId() +
18511851
" is overloaded for NS: " + nsId;

‎hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/async/RouterAsyncRpcClient.java‎

Lines changed: 14 additions & 123 deletions
Original file line numberDiff line numberDiff line change
@@ -179,8 +179,14 @@ public Object invokeMethod(
179179
namenodes.toString(), params);
180180
}
181181
threadLocalContext.transfer();
182-
invokeMethodAsync(nsid, ugi, (List<FederationNamenodeContext>) namenodes,
182+
RouterRpcFairnessPolicyController controller = getRouterRpcFairnessPolicyController();
183+
acquirePermit(nsid, ugi, method.getName(), controller);
184+
invokeMethodAsync(ugi, (List<FederationNamenodeContext>) namenodes,
183185
useObserver, protocol, method, params);
186+
asyncFinally(object -> {
187+
releasePermit(nsid, ugi, method, controller);
188+
return object;
189+
});
184190
}, router.getRpcServer().getAsyncRouterHandlerExecutors().getOrDefault(nsid,
185191
router.getRpcServer().getRouterAsyncHandlerDefaultExecutor()));
186192
return null;
@@ -203,13 +209,11 @@ public Object invokeMethod(
203209
* @param params The parameters for the method invocation.
204210
*/
205211
private void invokeMethodAsync(
206-
String nsid,
207212
final UserGroupInformation ugi,
208213
final List<FederationNamenodeContext> namenodes,
209214
boolean useObserver,
210215
final Class<?> protocol, final Method method, final Object... params) {
211216

212-
RouterRpcFairnessPolicyController controller = getRouterRpcFairnessPolicyController();
213217
addClientInfoToCallerContext(ugi);
214218
if (rpcMonitor != null) {
215219
rpcMonitor.proxyOp();
@@ -218,7 +222,6 @@ private void invokeMethodAsync(
218222
Map<FederationNamenodeContext, IOException> ioes = new LinkedHashMap<>();
219223
final ConnectionContext[] connection = new ConnectionContext[1];
220224
asyncTry(() -> {
221-
acquirePermit(nsid, ugi, method, controller);
222225
asyncForEach(namenodes.iterator(),
223226
(foreach, namenode) -> {
224227
if (!status.isShouldUseObserver()
@@ -260,12 +263,6 @@ private void invokeMethodAsync(
260263
return handlerAllNamenodeFail(namenodes, method, ioes, params);
261264
});
262265
});
263-
264-
asyncFinally(res -> {
265-
releasePermit(nsid, ugi, method, controller);
266-
return res;
267-
});
268-
269266
}
270267

271268
/**
@@ -429,9 +426,6 @@ public <R extends RemoteLocationContext, T> RemoteResult invokeSequential(
429426
}
430427
return ret;
431428
}, Exception.class);
432-
asyncFinally(ret -> {
433-
return ret;
434-
});
435429
});
436430
asyncApply(result -> {
437431
if (status.isComplete()) {
@@ -488,76 +482,6 @@ public <T extends RemoteLocationContext, R> Map<T, R> invokeConcurrent(
488482
return asyncReturn(Map.class);
489483
}
490484

491-
@SuppressWarnings("unchecked")
492-
public <T extends RemoteLocationContext, R> List<RemoteResult<T, R>> invokeConcurrent(
493-
final Collection<T> locations, final RemoteMethod method,
494-
boolean standby, long timeOutMs,
495-
Class<R> clazz) throws IOException {
496-
497-
final UserGroupInformation ugi = RouterRpcServer.getRemoteUser();
498-
final Method m = method.getMethod();
499-
500-
if (locations.isEmpty()) {
501-
throw new IOException("No remote locations available");
502-
} else if (locations.size() == 1 && timeOutMs <= 0) {
503-
// Shortcut, just one call
504-
return invokeSingle(locations.iterator().next(), method);
505-
}
506-
// Don't acquire CONCURRENT_NS permit here.
507-
RouterRpcFairnessPolicyController controller = getRouterRpcFairnessPolicyController();
508-
509-
List<T> orderedLocations = new ArrayList<>();
510-
List<Callable<Object>> callables = new ArrayList<>();
511-
// transfer originCall & callerContext to worker threads of executor.
512-
final Server.Call originCall = Server.getCurCall().get();
513-
final CallerContext originContext = CallerContext.getCurrent();
514-
for (final T location : locations) {
515-
String nsId = location.getNameserviceId();
516-
boolean isObserverRead = isObserverReadEligible(nsId, m);
517-
final List<? extends FederationNamenodeContext> namenodes =
518-
getOrderedNamenodes(nsId, isObserverRead);
519-
final Class<?> proto = method.getProtocol();
520-
final Object[] paramList = method.getParams(location);
521-
if (standby) {
522-
// Call the objectGetter to all NNs (including standby)
523-
for (final FederationNamenodeContext nn : namenodes) {
524-
String nnId = nn.getNamenodeId();
525-
final List<FederationNamenodeContext> nnList =
526-
Collections.singletonList(nn);
527-
T nnLocation = location;
528-
if (location instanceof RemoteLocation) {
529-
nnLocation = (T)new RemoteLocation(nsId, nnId, location.getDest());
530-
}
531-
orderedLocations.add(nnLocation);
532-
callables.add(
533-
() -> {
534-
transferThreadLocalContext(originCall, originContext);
535-
return invokeMethod(
536-
ugi, nnList, isObserverRead, proto, m, paramList);
537-
});
538-
}
539-
} else {
540-
// Call the objectGetter in order of nameservices in the NS list
541-
orderedLocations.add(location);
542-
callables.add(
543-
() -> {
544-
transferThreadLocalContext(originCall, originContext);
545-
return invokeMethod(
546-
ugi, namenodes, isObserverRead, proto, m, paramList);
547-
});
548-
}
549-
}
550-
551-
if (rpcMonitor != null) {
552-
rpcMonitor.proxyOp();
553-
}
554-
if (this.router.getRouterClientMetrics() != null) {
555-
this.router.getRouterClientMetrics().incInvokedConcurrent(m);
556-
}
557-
558-
return getRemoteResults(method, timeOutMs, controller, orderedLocations, callables);
559-
}
560-
561485
/**
562486
* Invokes multiple concurrent proxy calls to different clients. Returns an
563487
* array of results.
@@ -641,9 +565,6 @@ public <T extends RemoteLocationContext, R> List<RemoteResult<T, R>> invokeSingl
641565
asyncCatch((o, ioe) -> {
642566
throw processException(ioe, location);
643567
}, IOException.class);
644-
asyncFinally(o -> {
645-
return o;
646-
});
647568
return asyncReturn(List.class);
648569
}
649570

@@ -662,18 +583,13 @@ public <T extends RemoteLocationContext, R> List<RemoteResult<T, R>> invokeSingl
662583
public Object invokeSingle(final String nsId, RemoteMethod method)
663584
throws IOException {
664585
UserGroupInformation ugi = RouterRpcServer.getRemoteUser();
665-
asyncTry(() -> {
666-
boolean isObserverRead = isObserverReadEligible(nsId, method.getMethod());
667-
List<? extends FederationNamenodeContext> nns = getOrderedNamenodes(nsId, isObserverRead);
668-
RemoteLocationContext loc = new RemoteLocation(nsId, "/", "/");
669-
Class<?> proto = method.getProtocol();
670-
Method m = method.getMethod();
671-
Object[] params = method.getParams(loc);
672-
invokeMethod(ugi, nns, isObserverRead, proto, m, params);
673-
});
674-
asyncFinally(o -> {
675-
return o;
676-
});
586+
boolean isObserverRead = isObserverReadEligible(nsId, method.getMethod());
587+
List<? extends FederationNamenodeContext> nns = getOrderedNamenodes(nsId, isObserverRead);
588+
RemoteLocationContext loc = new RemoteLocation(nsId, "/", "/");
589+
Class<?> proto = method.getProtocol();
590+
Method m = method.getMethod();
591+
Object[] params = method.getParams(loc);
592+
invokeMethod(ugi, nns, isObserverRead, proto, m, params);
677593
return null;
678594
}
679595

@@ -698,31 +614,6 @@ public <T> T invokeSingle(
698614
return asyncReturn(clazz);
699615
}
700616

701-
protected void acquirePermit(final String nsId, final UserGroupInformation ugi,
702-
final Method m, RouterRpcFairnessPolicyController controller)
703-
throws IOException {
704-
if (controller != null) {
705-
if (!controller.acquirePermit(nsId)) {
706-
// Throw StandByException,
707-
// Clients could fail over and try another router.
708-
if (rpcMonitor != null) {
709-
rpcMonitor.proxyOpPermitRejected(nsId);
710-
}
711-
incrRejectedPermitForNs(nsId);
712-
LOG.debug("Permit denied for ugi: {} for method: {}",
713-
ugi, m.getName());
714-
String msg =
715-
"Router " + router.getRouterId() +
716-
" is overloaded for NS: " + nsId;
717-
throw new StandbyException(msg);
718-
}
719-
if (rpcMonitor != null) {
720-
rpcMonitor.proxyOpPermitAccepted(nsId);
721-
}
722-
incrAcceptedPermitForNs(nsId);
723-
}
724-
}
725-
726617
/**
727618
* Release permit for specific nsId after processing against downstream
728619
* nsId is completed.

0 commit comments

Comments
 (0)