You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@helix.apache.org by jx...@apache.org on 2019/02/26 00:26:28 UTC
[helix] branch master updated (8199d08 -> 70f8b2c)
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git.
from 8199d08 Add quota-based scheduling to tutorial
new ac2b715 Implement quota based workflow queues for existing pipeline
new 70f8b2c Change format to resolve conflict
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/helix/common/caches/TaskDataCache.java | 6 +-
.../stages/task/TaskSchedulingStage.java | 148 ++++++++++++++-------
.../helix/task/AssignableInstanceManager.java | 14 +-
.../java/org/apache/helix/task/JobDispatcher.java | 4 +-
.../TestAutoRebalancePartitionLimit.java | 4 +-
.../helix/integration/task/TestTaskThrottling.java | 3 +-
6 files changed, 118 insertions(+), 61 deletions(-)
[helix] 02/02: Change format to resolve conflict
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit 70f8b2cda571db70d5cfdcce0eedc33f765124bb
Author: Junkai Xue <jx...@linkedin.com>
AuthorDate: Mon Feb 25 16:26:10 2019 -0800
Change format to resolve conflict
---
.../java/org/apache/helix/common/caches/TaskDataCache.java | 6 +++---
.../org/apache/helix/task/AssignableInstanceManager.java | 14 +++++++-------
2 files changed, 10 insertions(+), 10 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/common/caches/TaskDataCache.java b/helix-core/src/main/java/org/apache/helix/common/caches/TaskDataCache.java
index 2c42aca..285fbe9 100644
--- a/helix-core/src/main/java/org/apache/helix/common/caches/TaskDataCache.java
+++ b/helix-core/src/main/java/org/apache/helix/common/caches/TaskDataCache.java
@@ -220,8 +220,8 @@ public class TaskDataCache extends AbstractDataCache {
}
}
- boolean[] updateSuccess =
- accessor.getBaseDataAccessor().setChildren(contextUpdatePaths, contextUpdateData, AccessOption.PERSISTENT);
+ boolean[] updateSuccess = accessor.getBaseDataAccessor()
+ .setChildren(contextUpdatePaths, contextUpdateData, AccessOption.PERSISTENT);
for (int i = 0; i < updateSuccess.length; i++) {
if (updateSuccess[i]) {
@@ -288,4 +288,4 @@ public class TaskDataCache extends AbstractDataCache {
return String.format("/%s/%s%s/%s/%s", _clusterName, PropertyType.PROPERTYSTORE.name(),
TaskConstants.REBALANCER_CONTEXT_ROOT, resourceName, TaskConstants.CONTEXT_NODE);
}
-}
\ No newline at end of file
+}
diff --git a/helix-core/src/main/java/org/apache/helix/task/AssignableInstanceManager.java b/helix-core/src/main/java/org/apache/helix/task/AssignableInstanceManager.java
index 7ab3f0d..7a79760 100644
--- a/helix-core/src/main/java/org/apache/helix/task/AssignableInstanceManager.java
+++ b/helix-core/src/main/java/org/apache/helix/task/AssignableInstanceManager.java
@@ -274,13 +274,13 @@ public class AssignableInstanceManager {
return true;
}
switch (state) {
- case ABORTED:
- case FAILED:
- case STOPPED:
- case COMPLETED:
- case TIMED_OUT:
- case NOT_STARTED:
- return true;
+ case ABORTED:
+ case FAILED:
+ case STOPPED:
+ case COMPLETED:
+ case TIMED_OUT:
+ case NOT_STARTED:
+ return true;
}
return false;
}
[helix] 01/02: Implement quota based workflow queues for existing
pipeline
Posted by jx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jxue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/helix.git
commit ac2b715400f02c9735b21bc46bd7a7be51e789c5
Author: Junkai Xue <jx...@linkedin.com>
AuthorDate: Fri Oct 12 11:05:45 2018 -0700
Implement quota based workflow queues for existing pipeline
The first step for task assignment witb producer/consumer mode is to have quota based workflow queues. For this ticket, it tracks the implementation for these queues to hold the order of workflows.
---
.../stages/task/TaskSchedulingStage.java | 148 ++++++++++++++-------
.../java/org/apache/helix/task/JobDispatcher.java | 4 +-
.../TestAutoRebalancePartitionLimit.java | 4 +-
.../helix/integration/task/TestTaskThrottling.java | 3 +-
4 files changed, 108 insertions(+), 51 deletions(-)
diff --git a/helix-core/src/main/java/org/apache/helix/controller/stages/task/TaskSchedulingStage.java b/helix-core/src/main/java/org/apache/helix/controller/stages/task/TaskSchedulingStage.java
index fe52390..5399e17 100644
--- a/helix-core/src/main/java/org/apache/helix/controller/stages/task/TaskSchedulingStage.java
+++ b/helix-core/src/main/java/org/apache/helix/controller/stages/task/TaskSchedulingStage.java
@@ -1,6 +1,10 @@
package org.apache.helix.controller.stages.task;
+import com.google.common.collect.Maps;
+import java.util.ArrayList;
+import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.PriorityQueue;
import org.apache.helix.HelixManager;
@@ -20,18 +24,20 @@ import org.apache.helix.model.Partition;
import org.apache.helix.model.Resource;
import org.apache.helix.model.ResourceAssignment;
import org.apache.helix.monitoring.mbeans.ClusterStatusMonitor;
-import org.apache.helix.task.JobContext;
-import org.apache.helix.task.JobRebalancer;
import org.apache.helix.task.TaskConstants;
-import org.apache.helix.task.TaskDriver;
import org.apache.helix.task.TaskRebalancer;
+import org.apache.helix.task.WorkflowConfig;
import org.apache.helix.task.WorkflowContext;
+import org.apache.helix.task.WorkflowDispatcher;
+import org.apache.helix.task.assigner.AssignableInstance;
import org.apache.helix.util.HelixUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class TaskSchedulingStage extends AbstractBaseStage {
private static final Logger logger = LoggerFactory.getLogger(TaskSchedulingStage.class.getName());
+ private Map<String, PriorityQueue<WorkflowObject>> _quotaBasedWorkflowPQs = Maps.newHashMap();
+ private WorkflowDispatcher _workflowDispatcher;
@Override
public void process(ClusterEvent event) throws Exception {
@@ -50,6 +56,10 @@ public class TaskSchedulingStage extends AbstractBaseStage {
// Reset current INIT/RUNNING tasks on participants for throttling
cache.resetActiveTaskCount(currentStateOutput);
+ buildQuotaBasedWorkflowPQsAndInitDispatchers(cache,
+ (HelixManager) event.getAttribute(AttributeName.helixmanager.name()),
+ (ClusterStatusMonitor) event.getAttribute(AttributeName.clusterStatusMonitor.name()));
+
final BestPossibleStateOutput bestPossibleStateOutput =
compute(event, resourceMap, currentStateOutput);
event.addAttribute(AttributeName.BEST_POSSIBLE_STATE.name(), bestPossibleStateOutput);
@@ -59,27 +69,20 @@ public class TaskSchedulingStage extends AbstractBaseStage {
private BestPossibleStateOutput compute(ClusterEvent event, Map<String, Resource> resourceMap,
CurrentStateOutput currentStateOutput) {
ClusterDataCache cache = event.getAttribute(AttributeName.ClusterDataCache.name());
+ Map<String, Resource> restOfResources = new HashMap<>(resourceMap);
BestPossibleStateOutput output = new BestPossibleStateOutput();
+ final List<String> failureResources = new ArrayList<>();
+ // Queues only for Workflows
+ scheduleWorkflows(resourceMap, cache, restOfResources, failureResources);
- PriorityQueue<TaskSchedulingStage.ResourcePriority> resourcePriorityQueue =
- new PriorityQueue<>();
- TaskDriver taskDriver = null;
- HelixManager helixManager = event.getAttribute(AttributeName.helixmanager.name());
- if (helixManager != null) {
- taskDriver = new TaskDriver(helixManager);
- }
- for (Resource resource : resourceMap.values()) {
- resourcePriorityQueue.add(new TaskSchedulingStage.ResourcePriority(resource,
- cache.getIdealState(resource.getResourceName()), taskDriver));
- }
-
- // TODO: Replace this looping available resources with Workflow Queues
- for (Iterator<TaskSchedulingStage.ResourcePriority> itr = resourcePriorityQueue.iterator();
- itr.hasNext(); ) {
- Resource resource = itr.next().getResource();
+ // Current rest of resources including: jobs + only current state left over ones
+ Iterator<Resource> itr = restOfResources.values().iterator();
+ while (itr.hasNext()) {
+ Resource resource = itr.next();
if (!computeResourceBestPossibleState(event, cache, currentStateOutput, resource, output)) {
- LogUtil
- .logWarn(logger, _eventId, "Failed to assign tasks for " + resource.getResourceName());
+ failureResources.add(resource.getResourceName());
+ LogUtil.logWarn(logger, _eventId,
+ "Failed to calculate best possible states for " + resource.getResourceName());
}
}
@@ -155,15 +158,9 @@ public class TaskSchedulingStage extends AbstractBaseStage {
try {
HelixManager manager = event.getAttribute(AttributeName.helixmanager.name());
rebalancer.init(manager);
-
- // Use the internal MappingCalculator interface to compute the final assignment
- // The next release will support rebalancers that compute the mapping from start to finish
partitionStateAssignment = mappingCalculator
.computeBestPossiblePartitionState(cache, idealState, resource, currentStateOutput);
- for (Partition partition : resource.getPartitions()) {
- Map<String, String> newStateMap = partitionStateAssignment.getReplicaMap(partition);
- output.setState(resourceName, partition, newStateMap);
- }
+ updateOutput(resource, partitionStateAssignment, output);
// Check if calculation is done successfully
return true;
@@ -189,32 +186,89 @@ public class TaskSchedulingStage extends AbstractBaseStage {
return false;
}
- class ResourcePriority implements Comparable<ResourcePriority> {
- final Resource _resource;
- // By default, non-job resources and new jobs are assigned lowest priority
- Long _priority = Long.MAX_VALUE;
+ class WorkflowObject implements Comparable<WorkflowObject> {
+ String _workflowId;
+ Long _rankingValue;
+
+ public WorkflowObject(String workflowId, long rankingValue) {
+ _workflowId = workflowId;
+ _rankingValue = rankingValue;
+ }
- Resource getResource() {
- return _resource;
+ @Override
+ public int compareTo(WorkflowObject o) {
+ return (int) (_rankingValue - o._rankingValue);
}
+ }
- public ResourcePriority(Resource resource, IdealState idealState, TaskDriver taskDriver) {
- _resource = resource;
+ private void buildQuotaBasedWorkflowPQsAndInitDispatchers(ClusterDataCache cache, HelixManager manager,
+ ClusterStatusMonitor monitor) {
+ _quotaBasedWorkflowPQs.clear();
+ Map<String, String> quotaRatioMap = cache.getClusterConfig().getTaskQuotaRatioMap();
- if (taskDriver != null && idealState != null
- && idealState.getRebalancerClassName() != null
- && idealState.getRebalancerClassName().equals(JobRebalancer.class.getName())) {
- // Update priority for job resources, note that older jobs will be processed earlier
- JobContext jobContext = taskDriver.getJobContext(resource.getResourceName());
- if (jobContext != null && jobContext.getStartTime() != WorkflowContext.UNSTARTED) {
- _priority = jobContext.getStartTime();
- }
+ // Create quota based queues
+ if (quotaRatioMap == null || quotaRatioMap.size() == 0) {
+ _quotaBasedWorkflowPQs
+ .put(AssignableInstance.DEFAULT_QUOTA_TYPE, new PriorityQueue<WorkflowObject>());
+ } else {
+ for (String quotaType : quotaRatioMap.keySet()) {
+ _quotaBasedWorkflowPQs.put(quotaType, new PriorityQueue<WorkflowObject>());
}
}
- @Override
- public int compareTo(ResourcePriority otherJob) {
- return _priority.compareTo(otherJob._priority);
+ for (String workflowId : cache.getWorkflowConfigMap().keySet()) {
+ WorkflowConfig workflowConfig = cache.getWorkflowConfig(workflowId);
+ String workflowType = workflowConfig.getWorkflowType();
+ if (workflowType == null || !_quotaBasedWorkflowPQs.containsKey(workflowType)) {
+ workflowType = AssignableInstance.DEFAULT_QUOTA_TYPE;
+ }
+ // TODO: We can support customized sorting field for user. Currently sort by creation time
+ _quotaBasedWorkflowPQs.get(workflowType)
+ .add(new WorkflowObject(workflowId, workflowConfig.getRecord().getCreationTime()));
+ }
+ if (_workflowDispatcher == null) {
+ _workflowDispatcher = new WorkflowDispatcher();
+ }
+ _workflowDispatcher.init(manager);
+ _workflowDispatcher.setClusterStatusMonitor(monitor);
+ _workflowDispatcher.updateCache(cache.getTaskDataCache());
+ }
+
+ private void updateOutput(Resource resource, ResourceAssignment partitionStateAssignment,
+ BestPossibleStateOutput output) {
+ // Use the internal MappingCalculator interface to compute the final assignment
+ // The next release will support rebalancers that compute the mapping from start to finish
+ for (Partition partition : resource.getPartitions()) {
+ Map<String, String> newStateMap = partitionStateAssignment.getReplicaMap(partition);
+ output.setState(resource.getResourceName(), partition, newStateMap);
+ }
+ }
+
+ private void scheduleWorkflows(Map<String, Resource> resourceMap, ClusterDataCache cache,
+ Map<String, Resource> restOfResources, List<String> failureResources) {
+ for (PriorityQueue<WorkflowObject> quotaBasedWorkflowPQ : _quotaBasedWorkflowPQs.values()) {
+ Iterator<WorkflowObject> it = quotaBasedWorkflowPQ.iterator();
+ while (it.hasNext()) {
+ String workflowId = it.next()._workflowId;
+ Resource resource = resourceMap.get(workflowId);
+ // TODO : Resource is null could be workflow just created without any IdealState.
+ // Let's remove this check when Helix is independent from IdealState
+ if (resource != null) {
+ try {
+ WorkflowContext context = _workflowDispatcher
+ .getOrInitializeWorkflowContext(workflowId, cache.getTaskDataCache());
+ _workflowDispatcher
+ .updateWorkflowStatus(workflowId, cache.getWorkflowConfig(workflowId), context);
+ _workflowDispatcher
+ .assignWorkflow(workflowId, cache.getWorkflowConfig(workflowId), context);
+ restOfResources.remove(workflowId);
+ } catch (Exception e) {
+ LogUtil.logError(logger, _eventId,
+ "Error computing assignment for Workflow " + workflowId + ". Skipping.", e);
+ failureResources.add(workflowId);
+ }
+ }
+ }
}
}
}
diff --git a/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java b/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java
index 05d8e5c..0f2558c 100644
--- a/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java
+++ b/helix-core/src/main/java/org/apache/helix/task/JobDispatcher.java
@@ -28,6 +28,8 @@ import org.slf4j.LoggerFactory;
public class JobDispatcher extends AbstractTaskDispatcher {
private static final Logger LOG = LoggerFactory.getLogger(JobDispatcher.class);
private ClusterDataCache _clusterDataCache;
+ private static final Set<TaskState> intemediateStates = new HashSet<>(
+ Arrays.asList(TaskState.IN_PROGRESS, TaskState.NOT_STARTED, TaskState.STOPPING, TaskState.STOPPED));
public void updateCache(ClusterDataCache cache) {
_clusterDataCache = cache;
@@ -125,7 +127,7 @@ public class JobDispatcher extends AbstractTaskDispatcher {
jobState = workflowCtx.getJobState(jobName);
workflowState = workflowCtx.getWorkflowState();
- if (jobState == TaskState.IN_PROGRESS && (isTimeout(jobCtx.getStartTime(), jobCfg.getTimeout())
+ if (intemediateStates.contains(jobState) && (isTimeout(jobCtx.getStartTime(), jobCfg.getTimeout())
|| TaskState.TIMED_OUT.equals(workflowState))) {
jobState = TaskState.TIMING_OUT;
workflowCtx.setJobState(jobName, TaskState.TIMING_OUT);
diff --git a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestAutoRebalancePartitionLimit.java b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestAutoRebalancePartitionLimit.java
index 99fa0ec..df4e5ba 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestAutoRebalancePartitionLimit.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/rebalancer/TestAutoRebalancePartitionLimit.java
@@ -118,9 +118,9 @@ public class TestAutoRebalancePartitionLimit extends ZkStandAloneCMTestBase {
_participants[1].syncStop();
- // verifyBalanceExternalView();
+ // verifyBalanziceExternalView();
result =
- ClusterStateVerifier.verifyByZkCallback(new ExternalViewBalancedVerifier(_gZkClient,
+ ClusterStateVerifier.verifyByPolling(new ExternalViewBalancedVerifier(_gZkClient,
CLUSTER_NAME, TEST_DB));
Assert.assertTrue(result);
ev = manager.getHelixDataAccessor().getProperty(accessor.keyBuilder().externalView(TEST_DB));
diff --git a/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskThrottling.java b/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskThrottling.java
index 592feda..5dc698f 100644
--- a/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskThrottling.java
+++ b/helix-core/src/test/java/org/apache/helix/integration/task/TestTaskThrottling.java
@@ -102,7 +102,8 @@ public class TestTaskThrottling extends TaskTestBase {
TaskState.COMPLETED);
}
- @Test(dependsOnMethods = {"testTaskThrottle"})
+ // Disable this test since helix will have priority map when integrate with JobIterator.
+ @Test(dependsOnMethods = {"testTaskThrottle"}, enabled = false)
public void testJobPriority() throws InterruptedException {
int numTasks = 30 * _numNodes;
int perNodeTaskLimitation = 5;