myriad-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From smare...@apache.org
Subject incubator-myriad git commit: MYRIAD-157 Concurrency issues while trying to flexup and flexdown NMs…
Date Tue, 20 Oct 2015 00:48:02 GMT
Repository: incubator-myriad
Updated Branches:
  refs/heads/master fceaf7301 -> d29ee338e


MYRIAD-157 Concurrency issues while trying to flexup and flexdown NMs…

… at the same time

When someone tries to flexup and flex down multiple NMs at the same time there are multiple
issues that  surface up starting form ConcurrentModification Exception and ending with NodeTask
being null.

This closes: #12
Review: https://github.com/apache/incubator-myriad/pull/12


Project: http://git-wip-us.apache.org/repos/asf/incubator-myriad/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-myriad/commit/d29ee338
Tree: http://git-wip-us.apache.org/repos/asf/incubator-myriad/tree/d29ee338
Diff: http://git-wip-us.apache.org/repos/asf/incubator-myriad/diff/d29ee338

Branch: refs/heads/master
Commit: d29ee338e017673e6714e6f39eb02cc9e27fc116
Parents: fceaf73
Author: Yuliya Feldman <yfeldman@maprtech.com>
Authored: Mon Oct 19 17:46:31 2015 -0700
Committer: Santosh Marella <marella@gmail.com>
Committed: Mon Oct 19 17:46:31 2015 -0700

----------------------------------------------------------------------
 .../ebay/myriad/scheduler/MyriadOperations.java  | 11 +++++------
 .../ebay/myriad/scheduler/TaskTerminator.java    | 15 +++++++++++++--
 .../handlers/ResourceOffersEventHandler.java     | 13 +++++++++++++
 .../event/handlers/StatusUpdateEventHandler.java | 15 ++++++++++-----
 .../com/ebay/myriad/state/SchedulerState.java    | 19 +++++++++++++------
 5 files changed, 54 insertions(+), 19 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/d29ee338/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/MyriadOperations.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/MyriadOperations.java
b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/MyriadOperations.java
index aff604c..903120a 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/MyriadOperations.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/MyriadOperations.java
@@ -28,7 +28,6 @@ import com.ebay.myriad.scheduler.constraints.LikeConstraint;
 import com.ebay.myriad.state.NodeTask;
 import com.ebay.myriad.state.SchedulerState;
 import com.google.common.collect.Lists;
-import com.google.common.collect.Sets;
 import com.google.inject.Inject;
 
 import org.apache.mesos.Protos;
@@ -141,7 +140,7 @@ public class MyriadOperations {
       
       // Flex down Pending tasks, if any
       if (numScaledDown < numInstancesToScaleDown) {
-        Set<Protos.TaskID> pendingTasks = Sets.newHashSet(this.schedulerState.getPendingTaskIds(serviceName));
+        Collection<Protos.TaskID> pendingTasks = this.schedulerState.getPendingTaskIds(serviceName);
 
         for (Protos.TaskID taskId : pendingTasks) {
             this.schedulerState.makeTaskKillable(taskId);
@@ -155,7 +154,7 @@ public class MyriadOperations {
       
       // Flex down Staging tasks, if any
       if (numScaledDown < numInstancesToScaleDown) {
-          Set<Protos.TaskID> stagingTasks = Sets.newHashSet(this.schedulerState.getStagingTaskIds(serviceName));
+          Collection<Protos.TaskID> stagingTasks = this.schedulerState.getStagingTaskIds(serviceName);
 
           for (Protos.TaskID taskId : stagingTasks) {
               this.schedulerState.makeTaskKillable(taskId);
@@ -167,7 +166,7 @@ public class MyriadOperations {
       }
       int numStagingTasksScaledDown = numScaledDown - numPendingTasksScaledDown;
 
-      Set<NodeTask> activeTasks = Sets.newHashSet(this.schedulerState.getActiveTasksByType(serviceName));
+      Set<NodeTask> activeTasks = this.schedulerState.getActiveTasksByType(serviceName);
       if (numScaledDown < numInstancesToScaleDown) {
         for (NodeTask nodeTask : activeTasks) {
           this.schedulerState.makeTaskKillable(nodeTask.getTaskStatus().getTaskId());
@@ -191,7 +190,7 @@ public class MyriadOperations {
           profile, constraint, numInstancesToScaleDown) : 0;
     }
 
-  private int flexDownStagingTasks(ServiceResourceProfile profile, Constraint constraint,
int numInstancesToScaleDown) {
+    private int flexDownStagingTasks(ServiceResourceProfile profile, Constraint constraint,
int numInstancesToScaleDown) {
       return numInstancesToScaleDown > 0 ? flexDownTasks(schedulerState.getStagingTaskIDsForProfile(profile),
           profile, constraint, numInstancesToScaleDown) : 0;
     }
@@ -220,7 +219,7 @@ public class MyriadOperations {
         }
       }
       return numInstancesScaledDown;
-    }
+  }
 
   private boolean meetsConstraint(NodeTask nodeTask, Constraint constraint) {
     if (constraint != null) {

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/d29ee338/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/TaskTerminator.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/TaskTerminator.java
b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/TaskTerminator.java
index 9840099..a048f30 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/TaskTerminator.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/TaskTerminator.java
@@ -19,11 +19,15 @@
 package com.ebay.myriad.scheduler;
 
 import com.ebay.myriad.scheduler.fgs.OfferLifecycleManager;
+import com.ebay.myriad.state.NodeTask;
 import com.ebay.myriad.state.SchedulerState;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Sets;
+
 import java.util.Set;
+
 import javax.inject.Inject;
+
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.mesos.Protos.Status;
 import org.apache.mesos.Protos.TaskID;
@@ -71,8 +75,15 @@ public class TaskTerminator implements Runnable {
               this.schedulerState.removeTask(taskIdToKill);
             } else {
               Status status = this.driverManager.kill(taskIdToKill);
-              offerLifeCycleManager.declineOutstandingOffers(schedulerState.getTask(taskIdToKill).getHostname());
-              this.schedulerState.removeTask(taskIdToKill);
+              NodeTask task = schedulerState.getTask(taskIdToKill);
+              if (task != null) {
+                offerLifeCycleManager.declineOutstandingOffers(
+                    task.getHostname());
+                this.schedulerState.removeTask(taskIdToKill);
+              } else {
+                schedulerState.removeTask(taskIdToKill);
+                LOGGER.warn("NodeTask with taskId: {} does not exist", taskIdToKill);
+              }
               Preconditions.checkState(status == Status.DRIVER_RUNNING);
             }
         }

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/d29ee338/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
index 1ca9ec0..3abf9ab 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
@@ -30,6 +30,7 @@ import com.ebay.myriad.scheduler.event.ResourceOffersEvent;
 import com.ebay.myriad.scheduler.fgs.OfferLifecycleManager;
 import com.ebay.myriad.state.NodeTask;
 import com.ebay.myriad.state.SchedulerState;
+import com.google.common.collect.Sets;
 import com.lmax.disruptor.EventHandler;
 
 import java.util.Iterator;
@@ -111,11 +112,20 @@ public class ResourceOffersEventHandler implements EventHandler<ResourceOffersEv
         for (NodeTask nodeTask : nodeTasks) {
           nodeTask.setSlaveAttributes(offer.getAttributesList());
         }
+        // keep this in case SchedulerState gets out of sync. This should not happen with

+        // synchronizing addNodes method in SchedulerState
+        // but to keep it safe
+        final Set<Protos.TaskID> missingTasks = Sets.newHashSet();
         Set<Protos.TaskID> pendingTasks = schedulerState.getPendingTaskIds();
         if (CollectionUtils.isNotEmpty(pendingTasks)) {
           for (Protos.TaskID pendingTaskId : pendingTasks) {
             NodeTask taskToLaunch = schedulerState
                 .getTask(pendingTaskId);
+            if (taskToLaunch == null) {
+              missingTasks.add(pendingTaskId);
+              LOGGER.warn("Node task for TaskID: {} does not exist", pendingTaskId);
+              continue;
+            }
             String taskPrefix = taskToLaunch.getTaskPrefix();
             ServiceResourceProfile profile = taskToLaunch.getProfile();
             Constraint constraint = taskToLaunch.getConstraint();
@@ -154,6 +164,9 @@ public class ResourceOffersEventHandler implements EventHandler<ResourceOffersEv
               }
             }
           }
+          for (Protos.TaskID taskId : missingTasks) {
+            schedulerState.removeTask(taskId);
+          }
         }
       }
 

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/d29ee338/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/StatusUpdateEventHandler.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/StatusUpdateEventHandler.java
b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/StatusUpdateEventHandler.java
index bd46747..628f09e 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/StatusUpdateEventHandler.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/StatusUpdateEventHandler.java
@@ -20,9 +20,12 @@ package com.ebay.myriad.scheduler.event.handlers;
 
 import com.ebay.myriad.scheduler.event.StatusUpdateEvent;
 import com.ebay.myriad.scheduler.fgs.OfferLifecycleManager;
+import com.ebay.myriad.state.NodeTask;
 import com.ebay.myriad.state.SchedulerState;
 import com.lmax.disruptor.EventHandler;
+
 import javax.inject.Inject;
+
 import org.apache.mesos.Protos.TaskID;
 import org.apache.mesos.Protos.TaskState;
 import org.apache.mesos.Protos.TaskStatus;
@@ -52,8 +55,10 @@ public class StatusUpdateEventHandler implements EventHandler<StatusUpdateEvent>
         TaskStatus status = event.getStatus();
         this.schedulerState.updateTask(status);
         TaskID taskId = status.getTaskId();
-        if (!schedulerState.hasTask(taskId)) {
+        NodeTask task = schedulerState.getTask(taskId);
+        if (task == null) {
             LOGGER.warn("Task: {} not found, status: {}", taskId.getValue(), status.getState());
+            schedulerState.removeTask(taskId);
             return;
         }
         LOGGER.info("Status Update for task: {} | state: {}", taskId.getValue(), status.getState());
@@ -70,20 +75,20 @@ public class StatusUpdateEventHandler implements EventHandler<StatusUpdateEvent>
                 schedulerState.makeTaskActive(taskId);
                 break;
             case TASK_FINISHED:
-                offerLifecycleManager.declineOutstandingOffers(schedulerState.getTask(taskId).getHostname());
+                offerLifecycleManager.declineOutstandingOffers(task.getHostname());
                 schedulerState.removeTask(taskId);
                 break;
             case TASK_FAILED:
                 // Add to pending tasks
-              offerLifecycleManager.declineOutstandingOffers(schedulerState.getTask(taskId).getHostname());
+              offerLifecycleManager.declineOutstandingOffers(task.getHostname());
               schedulerState.makeTaskPending(taskId);
                 break;
             case TASK_KILLED:
-              offerLifecycleManager.declineOutstandingOffers(schedulerState.getTask(taskId).getHostname());
+              offerLifecycleManager.declineOutstandingOffers(task.getHostname());
                 schedulerState.removeTask(taskId);
                 break;
             case TASK_LOST:
-              offerLifecycleManager.declineOutstandingOffers(schedulerState.getTask(taskId).getHostname());
+              offerLifecycleManager.declineOutstandingOffers(task.getHostname());
                 schedulerState.makeTaskPending(taskId);
                 break;
             default:

http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/d29ee338/myriad-scheduler/src/main/java/com/ebay/myriad/state/SchedulerState.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/state/SchedulerState.java b/myriad-scheduler/src/main/java/com/ebay/myriad/state/SchedulerState.java
index f0ea778..08a4dfa 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/state/SchedulerState.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/state/SchedulerState.java
@@ -60,7 +60,12 @@ public class SchedulerState {
         loadStateStore();
     }
 
-    public void addNodes(Collection<NodeTask> nodes) {
+    /**
+     * Making method synchronized, so if someone tries flexup/down at the same time
+     * addNodes and removeTask will not put data into an inconsistent state
+     * @param nodes
+     */
+    public synchronized void addNodes(Collection<NodeTask> nodes) {
         if (CollectionUtils.isEmpty(nodes)) {
             LOGGER.info("No nodes to add");
             return;
@@ -453,11 +458,13 @@ public class SchedulerState {
 
     public SchedulerStateForType(String taskPrefix) {
       this.taskPrefix = taskPrefix;
-      this.pendingTasks = new HashSet<>();
-      this.stagingTasks = new HashSet<>();
-      this.activeTasks = new HashSet<>();
-      this.lostTasks = new HashSet<>();
-      this.killableTasks = new HashSet<>();
+      // Since Sets.newConcurrentHashSet is available only starting form Guava version 15
+      // and so far (Hadoop 2.7) uses guava 13 we can not easily use it
+      this.pendingTasks = Collections.newSetFromMap(new ConcurrentHashMap<Protos.TaskID,
Boolean>());
+      this.stagingTasks = Collections.newSetFromMap(new ConcurrentHashMap<Protos.TaskID,
Boolean>());
+      this.activeTasks = Collections.newSetFromMap(new ConcurrentHashMap<Protos.TaskID,
Boolean>());
+      this.lostTasks = Collections.newSetFromMap(new ConcurrentHashMap<Protos.TaskID,
Boolean>());
+      this.killableTasks = Collections.newSetFromMap(new ConcurrentHashMap<Protos.TaskID,
Boolean>());
 
     }
     @SuppressWarnings("unused")


Mime
View raw message