abstractdog commented on code in PR #6659: URL: https://github.com/apache/hive/pull/6659#discussion_r3812416947
########## llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapPluginBroker.java: ########## @@ -0,0 +1,140 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.hive.llap.tezplugins; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; + +import com.google.common.annotations.VisibleForTesting; + +/** + * Broker that pairs up the two LLAP-side Tez plugins. + * + * <p>{@link LlapTaskCommunicator} and {@link LlapTaskSchedulerService} are constructed + * independently by the Tez app-master and have to find each other so the communicator can call + * {@code scheduler.notifyStarted(taskAttemptID)} from the {@code SubmitWork} response callback. + * Tez offers no first-class way for the two to meet, so they do it here: whichever side + * initializes first for a given DAG parks itself in the appropriate map, and the second side + * finds it and pairs the two up. + * + * <p>The handshake is keyed on {@link ApplicationAttemptId} — both plugin contexts expose it, + * and both plugins for the same DAG share it. Keying on the attempt id confines each handshake + * to a single DAG so that, in JVMs hosting concurrent DAGs (MiniHS2, MiniLlapCluster, tests + * fanning out concurrent inserts), the communicator for DAG-A cannot accidentally pair with the + * scheduler for DAG-B. + * + * <p>A single JVM-wide {@link #INSTANCE} is shared by all plugin constructions. The pair-or-park + * decision is atomic under {@link #lock}, so the pairing side always observes a consistent view + * of the peer map. {@code shutdown()} on each plugin unconditionally calls the matching + * {@code unregister…} method so a partial DAG init (only one of the two plugins was constructed + * before something failed) does not leak an entry — entries are keyed on the never-reused + * {@link ApplicationAttemptId} and would otherwise sit in the map until the JVM died. + * + * <p>Production impact of this class is nil. A real Tez {@code DAGAppMaster} runs at most one DAG + * over its lifetime, so both maps hold at most one entry, put and removed at plugin construction + * / shutdown. Nothing on the task submission or scheduling hot path touches this class. + */ +final class LlapPluginBroker { + + static final LlapPluginBroker INSTANCE = new LlapPluginBroker(); + + /** + * Guards the atomic "look at the peer map, and either pair or park" section on both sides. + * Held for the tiny window of the two-branch check; the maps are also {@link ConcurrentMap}s + * so {@code unregister…} outside the pairing window is safe. + */ + private final Object lock = new Object(); + + private final ConcurrentMap<ApplicationAttemptId, LlapTaskCommunicator> pendingCommunicators = + new ConcurrentHashMap<>(); + private final ConcurrentMap<ApplicationAttemptId, LlapTaskSchedulerService> pendingSchedulers = + new ConcurrentHashMap<>(); Review Comment: good catch, indeed; fixed in https://github.com/apache/hive/pull/6659/changes/9616d7040c8a08c21af9fcbf3cd6c96e0ea5bbdc highly concurrent `TestLlapPluginBroker` passed after this change assuming your approval still holds with this modification -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
