SLIDER-302 role launch moves to async model
Project: http://git-wip-us.apache.org/repos/asf/incubator-slider/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-slider/commit/e75f605f Tree: http://git-wip-us.apache.org/repos/asf/incubator-slider/tree/e75f605f Diff: http://git-wip-us.apache.org/repos/asf/incubator-slider/diff/e75f605f Branch: refs/heads/feature/SLIDER-149_Support_a_YARN_service_registry Commit: e75f605fb569eacab9d8e9e449e8f764e9fec13e Parents: 3d02e1c Author: Steve Loughran <ste...@apache.org> Authored: Fri Aug 8 17:21:56 2014 +0100 Committer: Steve Loughran <ste...@apache.org> Committed: Fri Aug 8 17:21:56 2014 +0100 ---------------------------------------------------------------------- .../server/appmaster/RoleLaunchService.java | 18 +++++----- .../server/appmaster/SliderAppMaster.java | 3 +- .../server/appmaster/actions/QueueAccess.java | 35 ++++++++++++++++++++ .../server/appmaster/actions/QueueService.java | 4 ++- 4 files changed, 47 insertions(+), 13 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e75f605f/slider-core/src/main/java/org/apache/slider/server/appmaster/RoleLaunchService.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/RoleLaunchService.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/RoleLaunchService.java index b83c905..e8b6802 100644 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/RoleLaunchService.java +++ b/slider-core/src/main/java/org/apache/slider/server/appmaster/RoleLaunchService.java @@ -30,6 +30,7 @@ import org.apache.slider.providers.ProviderRole; import org.apache.slider.providers.ProviderService; import org.apache.slider.server.appmaster.actions.ActionStartContainer; import org.apache.slider.server.appmaster.actions.AsyncAction; +import org.apache.slider.server.appmaster.actions.QueueAccess; import org.apache.slider.server.appmaster.state.RoleInstance; import org.apache.slider.server.appmaster.state.RoleStatus; import org.apache.slider.server.services.workflow.WorkflowExecutorService; @@ -52,13 +53,11 @@ public class RoleLaunchService public static final String ROLE_LAUNCH_SERVICE = "RoleLaunchService"; + /** - * Callback to whatever has the task of actually running the container - * start operation + * Queue submission API */ - private final ContainerStartOperation containerStarter; - - private final Queue<AsyncAction> actionQueue; + private final QueueAccess actionQueue; /** * Provider bulding up the command @@ -93,15 +92,14 @@ public class RoleLaunchService * @param envVars environment variables * @param launcherTmpDirPath path for a temporary data in the launch process */ - public RoleLaunchService(ContainerStartOperation startOperation, - Queue<AsyncAction> actionQueue, ProviderService provider, + public RoleLaunchService(QueueAccess queueAccess, + ProviderService provider, SliderFileSystem fs, Path generatedConfDirPath, Map<String, String> envVars, Path launcherTmpDirPath) { super(ROLE_LAUNCH_SERVICE); - containerStarter = startOperation; - this.actionQueue = actionQueue; + this.actionQueue = queueAccess; this.fs = fs; this.generatedConfDirPath = generatedConfDirPath; this.launcherTmpDirPath = launcherTmpDirPath; @@ -216,7 +214,7 @@ public class RoleLaunchService instance.role = containerRole; instance.roleId = role.id; instance.environment = envDescription; - actionQueue.add(new ActionStartContainer("starting " + containerRole, + actionQueue.put(new ActionStartContainer("starting " + containerRole, 0, container, containerLauncher.completeContainerLaunch(), instance)); http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e75f605f/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java index b1ac6bb..b82967f 100644 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java +++ b/slider-core/src/main/java/org/apache/slider/server/appmaster/SliderAppMaster.java @@ -750,8 +750,7 @@ public class SliderAppMaster extends AbstractSliderLaunchedService fs.getFileSystem().mkdirs(launcherTmpDirPath); //launcher service - launchService = new RoleLaunchService(this, - actionQueues.actionQueue, + launchService = new RoleLaunchService(actionQueues, providerService, fs, new Path(getGeneratedConfDir()), http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e75f605f/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueAccess.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueAccess.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueAccess.java new file mode 100644 index 0000000..f717085 --- /dev/null +++ b/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueAccess.java @@ -0,0 +1,35 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.slider.server.appmaster.actions; + +public interface QueueAccess { + /** + * Put an action on the immediate queue -to be executed when the queue + * reaches it. + * @param action action to queue + */ + void put(AsyncAction action); + + /** + * Put a delayed action: this will only be added to the main queue + * after its action time has been reached + * @param action action to queue + */ + void putDelayed(AsyncAction action); +} http://git-wip-us.apache.org/repos/asf/incubator-slider/blob/e75f605f/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueService.java ---------------------------------------------------------------------- diff --git a/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueService.java b/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueService.java index 69cb9f7..9e4e8d0 100644 --- a/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueService.java +++ b/slider-core/src/main/java/org/apache/slider/server/appmaster/actions/QueueService.java @@ -28,7 +28,7 @@ import java.util.concurrent.DelayQueue; import java.util.concurrent.LinkedBlockingDeque; public class QueueService extends AbstractService - implements Runnable { + implements Runnable, QueueAccess { private static final Logger log = LoggerFactory.getLogger(QueueService.class); @@ -41,11 +41,13 @@ public class QueueService extends AbstractService super("action queue"); } + @Override public void put(AsyncAction action) { log.debug("Queueing {}", action); actionQueue.add(action); } + @Override public void putDelayed(AsyncAction action) { log.debug("Delayed Queueing {}", action); delayedActions.add(action);
