Merge pull request #2244 from HubSpot/task-lag-guardrail

Avoid Request Locks With Task Lag

Author: WH77

SingularityService/src/main/java/com/hubspot/singularity/data/history/SingularityDeployHistoryPersister.java

@@ -135,7 +135,8 @@ public void runActionOnPoll() {
             }
           },
           requestId,
-          getClass().getSimpleName()
+          getClass().getSimpleName(),
+          SingularitySchedulerLock.Priority.LOW
         );
       }
 

SingularityService/src/main/java/com/hubspot/singularity/data/history/SingularityRequestHistoryPersister.java

@@ -20,6 +20,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantLock;
 import javax.inject.Singleton;
 import org.slf4j.Logger;
@@ -167,7 +168,8 @@ public void runActionOnPoll() {
             }
           },
           requestHistoryParent.requestId,
-          "request history purger"
+          "request history purger",
+          SingularitySchedulerLock.Priority.LOW
         );
       }
 

SingularityService/src/main/java/com/hubspot/singularity/helpers/TaskLagGuardrail.java

@@ -0,0 +1,63 @@
+package com.hubspot.singularity.helpers;
+
+import com.google.inject.Inject;
+import com.hubspot.singularity.SingularityManagedScheduledExecutorServiceFactory;
+import com.hubspot.singularity.SingularityPendingTaskId;
+import com.hubspot.singularity.config.SingularityConfiguration;
+import com.hubspot.singularity.data.TaskManager;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class TaskLagGuardrail {
+  private final TaskManager taskManager;
+  private final SingularityConfiguration configuration;
+  private final ScheduledExecutorService executor;
+  private ConcurrentMap<String, Integer> lateTasksByRequestId;
+
+  @Inject
+  public TaskLagGuardrail(
+    SingularityConfiguration configuration,
+    SingularityManagedScheduledExecutorServiceFactory factory,
+    TaskManager taskManager
+  ) {
+    this.configuration = configuration;
+    this.taskManager = taskManager;
+    this.lateTasksByRequestId = new ConcurrentHashMap<>();
+    this.executor = factory.getSingleThreaded(getClass().getSimpleName());
+    executor.scheduleWithFixedDelay(
+      this::updateLateTasksByRequestId,
+      0,
+      configuration.getRequestCacheTtl(),
+      TimeUnit.SECONDS
+    );
+  }
+
+  public boolean isLagged(String requestId) {
+    return lateTasksByRequestId.containsKey(requestId);
+  }
+
+  public void updateLateTasksByRequestId() {
+    long now = System.currentTimeMillis();
+    List<SingularityPendingTaskId> allPendingTaskIds = taskManager.getPendingTaskIds();
+
+    // not a thread safe assignment, but should be fine for periodic updates
+    this.lateTasksByRequestId =
+      allPendingTaskIds
+        .stream()
+        .filter(
+          p ->
+            now - p.getNextRunAt() > configuration.getDeltaAfterWhichTasksAreLateMillis()
+        )
+        .collect(
+          Collectors.toConcurrentMap(
+            SingularityPendingTaskId::getRequestId,
+            p -> 1,
+            Integer::sum
+          )
+        );
+  }
+}

SingularityService/src/main/java/com/hubspot/singularity/mesos/SingularityMesosModule.java

@@ -9,6 +9,7 @@
 import com.google.inject.multibindings.Multibinder;
 import com.google.inject.name.Named;
 import com.hubspot.singularity.helpers.MesosProtosUtils;
+import com.hubspot.singularity.helpers.TaskLagGuardrail;
 import com.hubspot.singularity.hooks.DeployAcceptanceHook;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -34,6 +35,7 @@ public void configure() {
     bind(SingularityStartup.class).in(Scopes.SINGLETON);
     bind(SingularitySchedulerLock.class).in(Scopes.SINGLETON);
     bind(SingularityMesosSchedulerClient.class).in(Scopes.SINGLETON);
+    bind(TaskLagGuardrail.class).in(Scopes.SINGLETON);
 
     Multibinder.newSetBinder(binder(), DeployAcceptanceHook.class);
   }

SingularityService/src/main/java/com/hubspot/singularity/mesos/SingularitySchedulerLock.java

@@ -2,6 +2,7 @@
 
 import com.google.inject.Inject;
 import com.hubspot.mesos.JavaUtils;
+import com.hubspot.singularity.helpers.TaskLagGuardrail;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
@@ -18,9 +19,11 @@ public class SingularitySchedulerLock {
   private final ReentrantLock stateLock;
   private final ReentrantLock offersLock;
   private final ConcurrentHashMap<String, ReentrantLock> requestLocks;
+  private final TaskLagGuardrail taskLag;
 
   @Inject
-  public SingularitySchedulerLock() {
+  public SingularitySchedulerLock(TaskLagGuardrail taskLag) {
+    this.taskLag = taskLag;
     this.stateLock = new ReentrantLock();
     this.offersLock = new ReentrantLock();
     this.requestLocks = new ConcurrentHashMap<>();
@@ -96,7 +99,38 @@ private void unlock(String requestId, String name, long start) {
     lock.unlock();
   }
 
+  /**
+   * Run the given function with the specified request lock.
+   *
+   * @param function The function to run.
+   * @param requestId Request to lock.
+   * @param name Description of this request lock.
+   */
   public void runWithRequestLock(Runnable function, String requestId, String name) {
+    runWithRequestLock(function, requestId, name, Priority.HIGH);
+  }
+
+  /**
+   * Run the given function with the specified request lock, unless run with low priority.
+   * If run with low priority, the function will not run if the request is lagged
+   * to allow higher priority components to acquire the lock.
+   *
+   * @param function The function to run.
+   * @param requestId Request to lock.
+   * @param name Description of this request lock.
+   * @param priority Priority of this request lock.
+   */
+  public void runWithRequestLock(
+    Runnable function,
+    String requestId,
+    String name,
+    Priority priority
+  ) {
+    if (priority == Priority.LOW && isLocked(requestId) && taskLag.isLagged(requestId)) {
+      LOG.info("{} - Skipping low priority lock on {}", name, requestId);
+      return;
+    }
+
     long start = lock(requestId, name);
     try {
       function.run();
@@ -208,4 +242,14 @@ private void unlockOffers(String name, long start) {
     LOG.debug("{} - Unlocking offers lock ({})", name, JavaUtils.duration(start));
     offersLock.unlock();
   }
+
+  private boolean isLocked(String requestId) {
+    ReentrantLock lock = requestLocks.get(requestId);
+    return lock != null && lock.isLocked();
+  }
+
+  public enum Priority {
+    LOW,
+    HIGH
+  }
 }

SingularityService/src/main/java/com/hubspot/singularity/scheduler/SingularityAutoScaleSpreadAllPoller.java

@@ -87,7 +87,8 @@ public void runActionOnPoll() {
           }
         },
         requestWithState.getRequest().getId(),
-        getClass().getSimpleName()
+        getClass().getSimpleName(),
+        SingularitySchedulerLock.Priority.LOW
       );
     }
   }

SingularityService/src/main/java/com/hubspot/singularity/scheduler/SingularityCrashLoopChecker.java

@@ -83,7 +83,8 @@ public void checkCooldowns() {
                         }
                       },
                       cooldownRequest.getRequest().getId(),
-                      getClass().getSimpleName()
+                      getClass().getSimpleName(),
+                      SingularitySchedulerLock.Priority.LOW
                     ),
                   cooldownExecutor
                 )

SingularityService/src/main/java/com/hubspot/singularity/scheduler/SingularityExpiringUserActionPoller.java

@@ -199,7 +199,8 @@ protected void checkExpiringObjects() {
                 );
               },
               requestWithState.get().getRequest().getId(),
-              getClazz().getSimpleName()
+              getClazz().getSimpleName(),
+              SingularitySchedulerLock.Priority.LOW
             );
           }
 

SingularityService/src/main/java/com/hubspot/singularity/scheduler/SingularitySchedulerModule.java

@@ -4,6 +4,7 @@
 import com.google.inject.Scopes;
 import com.hubspot.singularity.data.history.SingularityHistoryPurger;
 import com.hubspot.singularity.helpers.RebalancingHelper;
+import com.hubspot.singularity.helpers.TaskLagGuardrail;
 import com.hubspot.singularity.mesos.SingularityMesosOfferScheduler;
 import com.hubspot.singularity.mesos.SingularityMesosTaskPrioritizer;
 

SingularityService/src/main/java/com/hubspot/singularity/scheduler/SingularityUpstreamChecker.java

@@ -355,7 +355,8 @@ public void syncUpstreams() {
       lock.runWithRequestLock(
         () -> syncUpstreamsForService(singularityRequest),
         singularityRequest.getId(),
-        getClass().getSimpleName()
+        getClass().getSimpleName(),
+        SingularitySchedulerLock.Priority.LOW
       );
     }
   }