Undumb the Drydock resource allocator pipeline

Summary:
This was the major goal of D3859/D3855, and to a lesser degree D3854/D3852.

As Drydock is allocating a resource, it may need to allocate other resources first. For example, if it's allocating a working copy, it may need to allocate a host first.

Currently, we have the process basically queue up the allocation (insert a task into the queue) and sleep() until it finishes. This is problematic for a bunch of reasons, but the major one is that if allocation takes more resources (host, port, machine, DNS) than you have daemons, they could all end up sleeping and waiting for some other daemon to do their work. This is really stupid. Even if you only take up some of them, you're spending slots sleeping when you could be doing useful work.

To partially get around this and make the CLI experience less dumb, there's this goofy `synchronous` flag that gets passed around everywhere and pushes the workflow through a pile of special cases. Basically the `synchronous` flag causes us to do everything in-process. But this is dumb too because we'd rather do things in parallel if we can, and we have to have a lot of special case code to make it work at all.

Get rid of all of this. Instead of sleep()ing, try to work on the tasks that need to be worked on. If another daemon grabbed them already that's fine, but in the worst case we just gracefully degrade and do everything in process. So we get the best of both worlds: if we have parallelizable tasks and free daemons, things will execute in parallel. If we have nonparallelizable tasks or no free daemons, things will execute in process.

Test Plan: Ran `drydock_control.php --trace` and saw it perform cascading allocations without sleeping or special casing.

Reviewers: btrahan

Reviewed By: btrahan

CC: aran

Maniphest Tasks: T2015

Differential Revision: https://secure.phabricator.com/D3861

Author: epriestley

resources/sql/patches/drydocktaskid.sql

@@ -0,0 +1 @@
+ALTER TABLE {$NAMESPACE}_drydock.drydock_lease ADD taskID INT UNSIGNED;

scripts/drydock/drydock_control.php

@@ -20,10 +20,11 @@
 $root = dirname(dirname(dirname(__FILE__)));
 require_once $root.'/scripts/__init_script__.php';
 
-PhutilServiceProfiler::installEchoListener();
+$args = new PhutilArgumentParser($argv);
+$args->parseStandardArguments();
+$args->parse(array());
 
 $allocator = new DrydockAllocator();
-$allocator->makeSynchronous();
 $allocator->setResourceType('webroot');
 $lease = $allocator->allocate();
 
@@ -46,5 +47,5 @@
 $lease->release();
 
 
-//$i_http = $lease->getInterface('httpd');
-//echo $i_http->getURI('/index.html')."\n";
+// $i_http = $lease->getInterface('httpd');
+// echo $i_http->getURI('/index.html')."\n";

src/applications/drydock/allocator/DrydockAllocator.php

@@ -20,7 +20,6 @@ final class DrydockAllocator {
 
   private $resourceType;
   private $lease;
-  private $synchronous;
 
   public function setResourceType($resource_type) {
     $this->resourceType = $resource_type;
@@ -31,10 +30,6 @@ public function getResourceType() {
     return $this->resourceType;
   }
 
-  public function makeSynchronous() {
-    $this->synchronous = true;
-  }
-
   public function getPendingLease() {
     if (!$this->lease) {
       $lease = new DrydockLease();
@@ -54,13 +49,9 @@ public function allocate() {
       'lease'       => $lease->getID(),
     );
 
-    if ($this->synchronous) {
-      $data['synchronous'] = true;
-      $worker = new DrydockAllocatorWorker($data);
-      $worker->executeTask();
-    } else {
-      PhabricatorWorker::scheduleTask('DrydockAllocatorWorker', $data);
-    }
+    $task = PhabricatorWorker::scheduleTask('DrydockAllocatorWorker', $data);
+
+    $lease->setTaskID($task->getID());
 
     return $lease;
   }

src/applications/drydock/allocator/DrydockAllocatorWorker.php

@@ -64,11 +64,6 @@ protected function doWork() {
       shuffle($blueprints);
 
       $blueprint = head($blueprints);
-
-      if (isset($data['synchronous'])) {
-        $blueprint->makeSynchronous();
-      }
-
       $resource = $blueprint->allocateResource();
     }
 

src/applications/drydock/blueprint/DrydockBlueprint.php

@@ -20,11 +20,6 @@ abstract class DrydockBlueprint {
 
   private $activeLease;
   private $activeResource;
-  private $synchronous;
-
-  final public function makeSynchronous() {
-    $this->synchronous = true;
-  }
 
   abstract public function getType();
   abstract public function getInterface(
@@ -40,9 +35,6 @@ protected function executeAcquireLease(
 
   protected function getAllocator($type) {
     $allocator = new DrydockAllocator();
-    if ($this->synchronous) {
-      $allocator->makeSynchronous();
-    }
     $allocator->setResourceType($type);
 
     return $allocator;

src/applications/drydock/storage/DrydockLease.php

@@ -24,6 +24,7 @@ final class DrydockLease extends DrydockDAO {
   protected $ownerPHID;
   protected $attributes = array();
   protected $status;
+  protected $taskID;
 
   private $resource;
 
@@ -95,25 +96,43 @@ private function assertActive() {
     }
   }
 
-  public function waitUntilActive() {
-    $this->reload();
+  public static function waitForLeases(array $leases) {
+    assert_instances_of($leases, 'DrydockLease');
+
+    $task_ids = array_filter(mpull($leases, 'getTaskID'));
+    PhabricatorWorker::waitForTasks($task_ids);
+
+    $unresolved = $leases;
     while (true) {
-      switch ($this->status) {
-        case DrydockLeaseStatus::STATUS_ACTIVE:
-          break 2;
-        case DrydockLeaseStatus::STATUS_RELEASED:
-        case DrydockLeaseStatus::STATUS_EXPIRED:
-        case DrydockLeaseStatus::STATUS_BROKEN:
-          throw new Exception("Lease will never become active!");
-        case DrydockLeaseStatus::STATUS_PENDING:
-          break;
+      foreach ($unresolved as $key => $lease) {
+        $lease->reload();
+        switch ($lease->getStatus()) {
+          case DrydockLeaseStatus::STATUS_ACTIVE:
+            unset($unresolved[$key]);
+            break;
+          case DrydockLeaseStatus::STATUS_RELEASED:
+          case DrydockLeaseStatus::STATUS_EXPIRED:
+          case DrydockLeaseStatus::STATUS_BROKEN:
+            throw new Exception("Lease will never become active!");
+          case DrydockLeaseStatus::STATUS_PENDING:
+            break;
+        }
+      }
+
+      if ($unresolved) {
+        sleep(1);
+      } else {
+        break;
       }
-      sleep(2);
-      $this->reload();
     }
 
-    $this->attachResource($this->loadResource());
+    foreach ($leases as $lease) {
+      $lease->attachResource($lease->loadResource());
+    }
+  }
 
+  public function waitUntilActive() {
+    self::waitForLeases(array($this));
     return $this;
   }
 

src/infrastructure/daemon/workers/PhabricatorWorker.php

@@ -107,4 +107,47 @@ final public static function scheduleTask($task_class, $data) {
       ->save();
   }
 
+
+  /**
+   * Wait for tasks to complete. If tasks are not leased by other workers, they
+   * will be executed in this process while waiting.
+   *
+   * @param list<int>   List of queued task IDs to wait for.
+   * @return void
+   */
+  final public static function waitForTasks(array $task_ids) {
+    $task_table = new PhabricatorWorkerActiveTask();
+
+    $waiting = array_combine($task_ids, $task_ids);
+    while ($waiting) {
+      $conn_w = $task_table->establishConnection('w');
+
+      // Check if any of the tasks we're waiting on are still queued. If they
+      // are not, we're done waiting.
+      $row = queryfx_one(
+        $conn_w,
+        'SELECT COUNT(*) N FROM %T WHERE id IN (%Ld)',
+        $task_table->getTableName(),
+        $waiting);
+      if (!$row['N']) {
+        // Nothing is queued anymore. Stop waiting.
+        break;
+      }
+
+      $tasks = id(new PhabricatorWorkerLeaseQuery())
+        ->withIDs($waiting)
+        ->setLimit(1)
+        ->execute();
+
+      if (!$tasks) {
+        // We were not successful in leasing anything. Sleep for a bit and
+        // see if we have better luck later.
+        sleep(1);
+        continue;
+      }
+
+      $task = head($tasks)->executeTask();
+    }
+  }
+
 }

src/infrastructure/daemon/workers/query/PhabricatorWorkerLeaseQuery.php

@@ -25,6 +25,7 @@ final class PhabricatorWorkerLeaseQuery extends PhabricatorQuery {
 
   const PHASE_UNLEASED = 'unleased';
   const PHASE_EXPIRED  = 'expired';
+  const PHASE_SELECT   = 'select';
 
   const DEFAULT_LEASE_DURATION = 60; // Seconds
 
@@ -66,7 +67,8 @@ public function execute() {
     foreach ($phases as $phase) {
       queryfx(
         $conn_w,
-        'UPDATE %T SET leaseOwner = %s, leaseExpires = UNIX_TIMESTAMP() + %d
+        'UPDATE %T task
+          SET leaseOwner = %s, leaseExpires = UNIX_TIMESTAMP() + %d
           %Q %Q %Q',
         $task_table->getTableName(),
         $lease_ownership_name,
@@ -90,11 +92,10 @@ public function execute() {
       'SELECT task.*, taskdata.data _taskData, UNIX_TIMESTAMP() _serverTime
         FROM %T task LEFT JOIN %T taskdata
           ON taskdata.id = task.dataID
-        WHERE leaseOwner = %s AND leaseExpires > UNIX_TIMESTAMP()
-        %Q %Q',
+        %Q %Q %Q',
       $task_table->getTableName(),
       $taskdata_table->getTableName(),
-      $lease_ownership_name,
+      $this->buildWhereClause($conn_w, self::PHASE_SELECT),
       $this->buildOrderClause($conn_w),
       $this->buildLimitClause($conn_w, $limit));
 
@@ -124,14 +125,21 @@ private function buildWhereClause(AphrontDatabaseConnection $conn_w, $phase) {
       case self::PHASE_EXPIRED:
         $where[] = 'leaseExpires < UNIX_TIMESTAMP()';
         break;
+      case self::PHASE_SELECT:
+        $where[] = qsprintf(
+          $conn_w,
+          'leaseOwner = %s',
+          $this->getLeaseOwnershipName());
+        $where[] = 'leaseExpires > UNIX_TIMESTAMP()';
+        break;
       default:
         throw new Exception("Unknown phase '{$phase}'!");
     }
 
     if ($this->ids) {
       $where[] = qsprintf(
         $conn_w,
-        'id IN (%Ld)',
+        'task.id IN (%Ld)',
         $this->ids);
     }
 

src/infrastructure/storage/patch/PhabricatorBuiltinPatchList.php

@@ -1020,6 +1020,10 @@ public function getPatches() {
         'type'    => 'sql',
         'name'    => $this->getPatchPath('daemontaskarchive.sql'),
       ),
+      'drydocktaskid.sql' => array(
+        'type'    => 'sql',
+        'name'    => $this->getPatchPath('drydocktaskid.sql'),
+      ),
     );
   }