diff --git a/resources/sql/autopatches/20190220.daemon_worker.completed.01.sql b/resources/sql/autopatches/20190220.daemon_worker.completed.01.sql
new file mode 100644
index 0000000000..37f5a89bba
--- /dev/null
+++ b/resources/sql/autopatches/20190220.daemon_worker.completed.01.sql
@@ -0,0 +1,2 @@
+ALTER TABLE {$NAMESPACE}_worker.worker_archivetask
+ ADD archivedEpoch INT UNSIGNED NULL;
diff --git a/resources/sql/autopatches/20190220.daemon_worker.completed.02.sql b/resources/sql/autopatches/20190220.daemon_worker.completed.02.sql
new file mode 100644
index 0000000000..f0040576a9
--- /dev/null
+++ b/resources/sql/autopatches/20190220.daemon_worker.completed.02.sql
@@ -0,0 +1,3 @@
+ALTER TABLE {$NAMESPACE}_worker.worker_activetask
+ ADD dateCreated int unsigned NOT NULL,
+ ADD dateModified int unsigned NOT NULL;
diff --git a/src/applications/daemon/controller/PhabricatorDaemonConsoleController.php b/src/applications/daemon/controller/PhabricatorDaemonConsoleController.php
index a70cde04c4..421008082f 100644
--- a/src/applications/daemon/controller/PhabricatorDaemonConsoleController.php
+++ b/src/applications/daemon/controller/PhabricatorDaemonConsoleController.php
@@ -1,269 +1,290 @@
<?php
final class PhabricatorDaemonConsoleController
extends PhabricatorDaemonController {
public function handleRequest(AphrontRequest $request) {
$viewer = $this->getViewer();
$window_start = (time() - (60 * 15));
// Assume daemons spend about 250ms second in overhead per task acquiring
// leases and doing other bookkeeping. This is probably an over-estimation,
// but we'd rather show that utilization is too high than too low.
$lease_overhead = 0.250;
$completed = id(new PhabricatorWorkerArchiveTaskQuery())
->withDateModifiedSince($window_start)
->execute();
$failed = id(new PhabricatorWorkerActiveTask())->loadAllWhere(
'failureTime > %d',
$window_start);
$usage_total = 0;
$usage_start = PHP_INT_MAX;
$completed_info = array();
foreach ($completed as $completed_task) {
$class = $completed_task->getTaskClass();
if (empty($completed_info[$class])) {
$completed_info[$class] = array(
'n' => 0,
'duration' => 0,
+ 'queueTime' => 0,
);
}
$completed_info[$class]['n']++;
$duration = $completed_task->getDuration();
$completed_info[$class]['duration'] += $duration;
// NOTE: Duration is in microseconds, but we're just using seconds to
// compute utilization.
$usage_total += $lease_overhead + ($duration / 1000000);
$usage_start = min($usage_start, $completed_task->getDateModified());
+
+ $date_archived = $completed_task->getArchivedEpoch();
+ $queue_seconds = $date_archived - $completed_task->getDateCreated();
+
+ // Don't measure queue time for tasks that completed in the same
+ // epoch-second they were created in.
+ if ($queue_seconds > 0) {
+ $sec_in_us = phutil_units('1 second in microseconds');
+ $queue_us = $queue_seconds * $sec_in_us;
+ $queue_exclusive_us = $queue_us - $duration;
+ $queue_exclusive_seconds = $queue_exclusive_us / $sec_in_us;
+ $rounded = floor($queue_exclusive_seconds);
+ $completed_info[$class]['queueTime'] += $rounded;
+ }
}
$completed_info = isort($completed_info, 'n');
$rows = array();
foreach ($completed_info as $class => $info) {
+ $duration_avg = new PhutilNumber((int)($info['duration'] / $info['n']));
+ $queue_avg = new PhutilNumber((int)($info['queueTime'] / $info['n']));
$rows[] = array(
$class,
number_format($info['n']),
- pht('%s us', new PhutilNumber((int)($info['duration'] / $info['n']))),
+ pht('%s us', $duration_avg),
+ pht('%s s', $queue_avg),
);
}
if ($failed) {
// Add the time it takes to restart the daemons. This includes a guess
// about other overhead of 2X.
$restart_delay = PhutilDaemonHandle::getWaitBeforeRestart();
$usage_total += $restart_delay * count($failed) * 2;
foreach ($failed as $failed_task) {
$usage_start = min($usage_start, $failed_task->getFailureTime());
}
$rows[] = array(
phutil_tag('em', array(), pht('Temporary Failures')),
count($failed),
null,
);
}
$logs = id(new PhabricatorDaemonLogQuery())
->setViewer($viewer)
->withStatus(PhabricatorDaemonLogQuery::STATUS_ALIVE)
->setAllowStatusWrites(true)
->execute();
$taskmasters = 0;
foreach ($logs as $log) {
if ($log->getDaemon() == 'PhabricatorTaskmasterDaemon') {
$taskmasters++;
}
}
if ($taskmasters && $usage_total) {
// Total number of wall-time seconds the daemons have been running since
// the oldest event. For very short times round up to 15s so we don't
// render any ridiculous numbers if you reload the page immediately after
// restarting the daemons.
$available_time = $taskmasters * max(15, (time() - $usage_start));
// Percentage of those wall-time seconds we can account for, which the
// daemons spent doing work:
$used_time = ($usage_total / $available_time);
$rows[] = array(
phutil_tag('em', array(), pht('Queue Utilization (Approximate)')),
sprintf('%.1f%%', 100 * $used_time),
null,
+ null,
);
}
$completed_table = new AphrontTableView($rows);
$completed_table->setNoDataString(
pht('No tasks have completed in the last 15 minutes.'));
$completed_table->setHeaders(
array(
pht('Class'),
pht('Count'),
- pht('Avg'),
+ pht('Average Duration'),
+ pht('Average Queue Time'),
));
$completed_table->setColumnClasses(
array(
'wide',
'n',
'n',
+ 'n',
));
$completed_panel = id(new PHUIObjectBoxView())
->setHeaderText(pht('Recently Completed Tasks (Last 15m)'))
->setTable($completed_table);
$daemon_table = id(new PhabricatorDaemonLogListView())
->setUser($viewer)
->setDaemonLogs($logs);
$daemon_panel = id(new PHUIObjectBoxView())
->setHeaderText(pht('Active Daemons'))
->setTable($daemon_table);
$tasks = id(new PhabricatorWorkerLeaseQuery())
->setSkipLease(true)
->withLeasedTasks(true)
->setLimit(100)
->execute();
$tasks_table = id(new PhabricatorDaemonTasksTableView())
->setTasks($tasks)
->setNoDataString(pht('No tasks are leased by workers.'));
$leased_panel = id(new PHUIObjectBoxView())
->setHeaderText(pht('Leased Tasks'))
->setTable($tasks_table);
$task_table = new PhabricatorWorkerActiveTask();
$queued = queryfx_all(
$task_table->establishConnection('r'),
'SELECT taskClass, count(*) N FROM %T GROUP BY taskClass
ORDER BY N DESC',
$task_table->getTableName());
$rows = array();
foreach ($queued as $row) {
$rows[] = array(
$row['taskClass'],
number_format($row['N']),
);
}
$queued_table = new AphrontTableView($rows);
$queued_table->setHeaders(
array(
pht('Class'),
pht('Count'),
));
$queued_table->setColumnClasses(
array(
'wide',
'n',
));
$queued_table->setNoDataString(pht('Task queue is empty.'));
$queued_panel = new PHUIObjectBoxView();
$queued_panel->setHeaderText(pht('Queued Tasks'));
$queued_panel->setTable($queued_table);
$upcoming = id(new PhabricatorWorkerLeaseQuery())
->setLimit(10)
->setSkipLease(true)
->execute();
$upcoming_panel = id(new PHUIObjectBoxView())
->setHeaderText(pht('Next In Queue'))
->setTable(
id(new PhabricatorDaemonTasksTableView())
->setTasks($upcoming)
->setNoDataString(pht('Task queue is empty.')));
$triggers = id(new PhabricatorWorkerTriggerQuery())
->setViewer($viewer)
->setOrder(PhabricatorWorkerTriggerQuery::ORDER_EXECUTION)
->withNextEventBetween(0, null)
->needEvents(true)
->setLimit(10)
->execute();
$triggers_table = $this->buildTriggersTable($triggers);
$triggers_panel = id(new PHUIObjectBoxView())
->setHeaderText(pht('Upcoming Triggers'))
->setTable($triggers_table);
$crumbs = $this->buildApplicationCrumbs();
$crumbs->addTextCrumb(pht('Console'));
$nav = $this->buildSideNavView();
$nav->selectFilter('/');
$nav->appendChild(
array(
$crumbs,
$completed_panel,
$daemon_panel,
$queued_panel,
$leased_panel,
$upcoming_panel,
$triggers_panel,
));
return $this->newPage()
->setTitle(pht('Console'))
->appendChild($nav);
}
private function buildTriggersTable(array $triggers) {
$viewer = $this->getViewer();
$rows = array();
foreach ($triggers as $trigger) {
$event = $trigger->getEvent();
if ($event) {
$last_epoch = $event->getLastEventEpoch();
$next_epoch = $event->getNextEventEpoch();
} else {
$last_epoch = null;
$next_epoch = null;
}
$rows[] = array(
$trigger->getID(),
$trigger->getClockClass(),
$trigger->getActionClass(),
$last_epoch ? phabricator_datetime($last_epoch, $viewer) : null,
$next_epoch ? phabricator_datetime($next_epoch, $viewer) : null,
);
}
return id(new AphrontTableView($rows))
->setNoDataString(pht('There are no upcoming event triggers.'))
->setHeaders(
array(
pht('ID'),
pht('Clock'),
pht('Action'),
pht('Last'),
pht('Next'),
))
->setColumnClasses(
array(
'',
'',
'wide',
'date',
'date',
));
}
}
diff --git a/src/infrastructure/daemon/workers/storage/PhabricatorWorkerActiveTask.php b/src/infrastructure/daemon/workers/storage/PhabricatorWorkerActiveTask.php
index b6bd462df7..7139e39ac3 100644
--- a/src/infrastructure/daemon/workers/storage/PhabricatorWorkerActiveTask.php
+++ b/src/infrastructure/daemon/workers/storage/PhabricatorWorkerActiveTask.php
@@ -1,214 +1,215 @@
<?php
final class PhabricatorWorkerActiveTask extends PhabricatorWorkerTask {
protected $failureTime;
private $serverTime;
private $localTime;
protected function getConfiguration() {
$parent = parent::getConfiguration();
$config = array(
self::CONFIG_IDS => self::IDS_COUNTER,
- self::CONFIG_TIMESTAMPS => false,
self::CONFIG_KEY_SCHEMA => array(
'taskClass' => array(
'columns' => array('taskClass'),
),
'leaseExpires' => array(
'columns' => array('leaseExpires'),
),
'key_failuretime' => array(
'columns' => array('failureTime'),
),
'key_owner' => array(
'columns' => array('leaseOwner', 'priority', 'id'),
),
) + $parent[self::CONFIG_KEY_SCHEMA],
);
return $config + $parent;
}
public function setServerTime($server_time) {
$this->serverTime = $server_time;
$this->localTime = time();
return $this;
}
public function setLeaseDuration($lease_duration) {
$this->checkLease();
$server_lease_expires = $this->serverTime + $lease_duration;
$this->setLeaseExpires($server_lease_expires);
// NOTE: This is primarily to allow unit tests to set negative lease
// durations so they don't have to wait around for leases to expire. We
// check that the lease is valid above.
return $this->forceSaveWithoutLease();
}
public function save() {
$this->checkLease();
return $this->forceSaveWithoutLease();
}
public function forceSaveWithoutLease() {
$is_new = !$this->getID();
if ($is_new) {
$this->failureCount = 0;
}
if ($is_new) {
$data = new PhabricatorWorkerTaskData();
$data->setData($this->getData());
$data->save();
$this->setDataID($data->getID());
}
return parent::save();
}
protected function checkLease() {
$owner = $this->leaseOwner;
if (!$owner) {
return;
}
if ($owner == PhabricatorWorker::YIELD_OWNER) {
return;
}
$current_server_time = $this->serverTime + (time() - $this->localTime);
if ($current_server_time >= $this->leaseExpires) {
throw new Exception(
pht(
'Trying to update Task %d (%s) after lease expiration!',
$this->getID(),
$this->getTaskClass()));
}
}
public function delete() {
throw new Exception(
pht(
'Active tasks can not be deleted directly. '.
'Use %s to move tasks to the archive.',
'archiveTask()'));
}
public function archiveTask($result, $duration) {
if ($this->getID() === null) {
throw new Exception(
pht("Attempting to archive a task which hasn't been saved!"));
}
$this->checkLease();
$archive = id(new PhabricatorWorkerArchiveTask())
->setID($this->getID())
->setTaskClass($this->getTaskClass())
->setLeaseOwner($this->getLeaseOwner())
->setLeaseExpires($this->getLeaseExpires())
->setFailureCount($this->getFailureCount())
->setDataID($this->getDataID())
->setPriority($this->getPriority())
->setObjectPHID($this->getObjectPHID())
->setResult($result)
- ->setDuration($duration);
+ ->setDuration($duration)
+ ->setDateCreated($this->getDateCreated())
+ ->setArchivedEpoch(PhabricatorTime::getNow());
// NOTE: This deletes the active task (this object)!
$archive->save();
return $archive;
}
public function executeTask() {
// We do this outside of the try .. catch because we don't have permission
// to release the lease otherwise.
$this->checkLease();
$did_succeed = false;
$worker = null;
try {
$worker = $this->getWorkerInstance();
$worker->setCurrentWorkerTask($this);
$maximum_failures = $worker->getMaximumRetryCount();
if ($maximum_failures !== null) {
if ($this->getFailureCount() > $maximum_failures) {
throw new PhabricatorWorkerPermanentFailureException(
pht(
'Task %d has exceeded the maximum number of failures (%d).',
$this->getID(),
$maximum_failures));
}
}
$lease = $worker->getRequiredLeaseTime();
if ($lease !== null) {
$this->setLeaseDuration($lease);
}
$t_start = microtime(true);
$worker->executeTask();
$duration = phutil_microseconds_since($t_start);
$result = $this->archiveTask(
PhabricatorWorkerArchiveTask::RESULT_SUCCESS,
$duration);
$did_succeed = true;
} catch (PhabricatorWorkerPermanentFailureException $ex) {
$result = $this->archiveTask(
PhabricatorWorkerArchiveTask::RESULT_FAILURE,
0);
$result->setExecutionException($ex);
} catch (PhabricatorWorkerYieldException $ex) {
$this->setExecutionException($ex);
$this->setLeaseOwner(PhabricatorWorker::YIELD_OWNER);
$retry = $ex->getDuration();
$retry = max($retry, 5);
// NOTE: As a side effect, this saves the object.
$this->setLeaseDuration($retry);
$result = $this;
} catch (Exception $ex) {
$this->setExecutionException($ex);
$this->setFailureCount($this->getFailureCount() + 1);
$this->setFailureTime(time());
$retry = null;
if ($worker) {
$retry = $worker->getWaitBeforeRetry($this);
}
$retry = coalesce(
$retry,
PhabricatorWorkerLeaseQuery::getDefaultWaitBeforeRetry());
// NOTE: As a side effect, this saves the object.
$this->setLeaseDuration($retry);
$result = $this;
}
// NOTE: If this throws, we don't want it to cause the task to fail again,
// so execute it out here and just let the exception escape.
if ($did_succeed) {
// Default the new task priority to our own priority.
$defaults = array(
'priority' => (int)$this->getPriority(),
);
$worker->flushTaskQueue($defaults);
}
return $result;
}
}
diff --git a/src/infrastructure/daemon/workers/storage/PhabricatorWorkerArchiveTask.php b/src/infrastructure/daemon/workers/storage/PhabricatorWorkerArchiveTask.php
index 0062d07a84..25a453b47b 100644
--- a/src/infrastructure/daemon/workers/storage/PhabricatorWorkerArchiveTask.php
+++ b/src/infrastructure/daemon/workers/storage/PhabricatorWorkerArchiveTask.php
@@ -1,97 +1,100 @@
<?php
final class PhabricatorWorkerArchiveTask extends PhabricatorWorkerTask {
const RESULT_SUCCESS = 0;
const RESULT_FAILURE = 1;
const RESULT_CANCELLED = 2;
protected $duration;
protected $result;
+ protected $archivedEpoch;
protected function getConfiguration() {
$parent = parent::getConfiguration();
$config = array(
// We manage the IDs in this table; they are allocated in the ActiveTask
// table and moved here without alteration.
self::CONFIG_IDS => self::IDS_MANUAL,
) + $parent;
$config[self::CONFIG_COLUMN_SCHEMA] = array(
'result' => 'uint32',
'duration' => 'uint64',
+ 'archivedEpoch' => 'epoch?',
) + $config[self::CONFIG_COLUMN_SCHEMA];
$config[self::CONFIG_KEY_SCHEMA] = array(
'dateCreated' => array(
'columns' => array('dateCreated'),
),
'key_modified' => array(
'columns' => array('dateModified'),
),
) + $parent[self::CONFIG_KEY_SCHEMA];
return $config;
}
public function save() {
if ($this->getID() === null) {
throw new Exception(pht('Trying to archive a task with no ID.'));
}
$other = new PhabricatorWorkerActiveTask();
$conn_w = $this->establishConnection('w');
$this->openTransaction();
queryfx(
$conn_w,
'DELETE FROM %T WHERE id = %d',
$other->getTableName(),
$this->getID());
$result = parent::insert();
$this->saveTransaction();
return $result;
}
public function delete() {
$this->openTransaction();
if ($this->getDataID()) {
$conn_w = $this->establishConnection('w');
$data_table = new PhabricatorWorkerTaskData();
queryfx(
$conn_w,
'DELETE FROM %T WHERE id = %d',
$data_table->getTableName(),
$this->getDataID());
}
$result = parent::delete();
$this->saveTransaction();
return $result;
}
public function unarchiveTask() {
$this->openTransaction();
$active = id(new PhabricatorWorkerActiveTask())
->setID($this->getID())
->setTaskClass($this->getTaskClass())
->setLeaseOwner(null)
->setLeaseExpires(0)
->setFailureCount(0)
->setDataID($this->getDataID())
->setPriority($this->getPriority())
->setObjectPHID($this->getObjectPHID())
+ ->setDateCreated($this->getDateCreated())
->insert();
$this->setDataID(null);
$this->delete();
$this->saveTransaction();
return $active;
}
}