Allow build to be bumped to the front of the queue via the web interface

Builds now have a "Bump up" action. This will cause the queue runner
to prioritise the steps of the build above all other steps.

src/hydra-queue-runner/dispatcher.cc

@@ -144,7 +144,9 @@ system_time State::doDispatch()
             {
                 auto a_(a->state.lock());
                 auto b_(b->state.lock()); // FIXME: deadlock?
-                return a_->lowestBuildID < b_->lowestBuildID;
+                return
+                    a_->highestGlobalPriority != b_->highestGlobalPriority ? a_->highestGlobalPriority > b_->highestGlobalPriority :
+                    a_->lowestBuildID < b_->lowestBuildID;
             });
 
         /* Find a machine with a free slot and find a step to run

src/hydra-queue-runner/queue-monitor.cc

@@ -25,6 +25,7 @@ void State::queueMonitorLoop()
     receiver buildsRestarted(*conn, "builds_restarted");
     receiver buildsCancelled(*conn, "builds_cancelled");
     receiver buildsDeleted(*conn, "builds_deleted");
+    receiver buildsBumped(*conn, "builds_bumped");
 
     auto store = openStore(); // FIXME: pool
 
@@ -44,9 +45,9 @@ void State::queueMonitorLoop()
             printMsg(lvlTalkative, "got notification: builds restarted");
             lastBuildId = 0; // check all builds
         }
-        if (buildsCancelled.get() || buildsDeleted.get()) {
-            printMsg(lvlTalkative, "got notification: builds cancelled");
-            removeCancelledBuilds(*conn);
+        if (buildsCancelled.get() || buildsDeleted.get() || buildsBumped.get()) {
+            printMsg(lvlTalkative, "got notification: builds cancelled or bumped");
+            processQueueChange(*conn);
         }
 
     }
@@ -64,7 +65,7 @@ void State::getQueuedBuilds(Connection & conn, std::shared_ptr<StoreAPI> store,
     {
         pqxx::work txn(conn);
 
-        auto res = txn.parameterized("select id, project, jobset, job, drvPath, maxsilent, timeout, timestamp from Builds where id > $1 and finished = 0 order by id")(lastBuildId).exec();
+        auto res = txn.parameterized("select id, project, jobset, job, drvPath, maxsilent, timeout, timestamp, globalPriority from Builds where id > $1 and finished = 0 order by id")(lastBuildId).exec();
 
         for (auto const & row : res) {
             auto builds_(builds.lock());
@@ -82,6 +83,7 @@ void State::getQueuedBuilds(Connection & conn, std::shared_ptr<StoreAPI> store,
             build->maxSilentTime = row["maxsilent"].as<int>();
             build->buildTimeout = row["timeout"].as<int>();
             build->timestamp = row["timestamp"].as<time_t>();
+            build->globalPriority = row["globalPriority"].as<int>();
 
             newBuilds.emplace(std::make_pair(build->drvPath, build));
         }
@@ -228,13 +230,7 @@ void State::getQueuedBuilds(Connection & conn, std::shared_ptr<StoreAPI> store,
             throw;
         }
 
-        /* Update the lowest build ID field of each dependency. This
-           is used by the dispatcher to start steps in order of build
-           ID. */
-        visitDependencies([&](const Step::ptr & step) {
-            auto step_(step->state.lock());
-            step_->lowestBuildID = std::min(step_->lowestBuildID, build->id);
-        }, build->toplevel);
+        build->propagatePriorities();
 
         /* Add the new runnable build steps to ‘runnable’ and wake up
            the builder threads. */
@@ -247,26 +243,47 @@ void State::getQueuedBuilds(Connection & conn, std::shared_ptr<StoreAPI> store,
 }
 
 
-void State::removeCancelledBuilds(Connection & conn)
+void Build::propagatePriorities()
+{
+    /* Update the highest global priority and lowest build ID fields
+       of each dependency. This is used by the dispatcher to start
+       steps in order of descending global priority and ascending
+       build ID. */
+    visitDependencies([&](const Step::ptr & step) {
+        auto step_(step->state.lock());
+        step_->highestGlobalPriority = std::max(step_->highestGlobalPriority, globalPriority);
+        step_->lowestBuildID = std::min(step_->lowestBuildID, id);
+    }, toplevel);
+}
+
+
+void State::processQueueChange(Connection & conn)
 {
     /* Get the current set of queued builds. */
-    std::set<BuildID> currentIds;
+    std::map<BuildID, int> currentIds;
     {
         pqxx::work txn(conn);
-        auto res = txn.exec("select id from Builds where finished = 0");
+        auto res = txn.exec("select id, globalPriority from Builds where finished = 0");
         for (auto const & row : res)
-            currentIds.insert(row["id"].as<BuildID>());
+            currentIds[row["id"].as<BuildID>()] = row["globalPriority"].as<BuildID>();
     }
 
     auto builds_(builds.lock());
 
     for (auto i = builds_->begin(); i != builds_->end(); ) {
-        if (currentIds.find(i->first) == currentIds.end()) {
+        auto b = currentIds.find(i->first);
+        if (b == currentIds.end()) {
             printMsg(lvlInfo, format("discarding cancelled build %1%") % i->first);
             i = builds_->erase(i);
             // FIXME: ideally we would interrupt active build steps here.
-        } else
-            ++i;
+            continue;
+        }
+        if (i->second->globalPriority < b->second) {
+            printMsg(lvlInfo, format("priority of build %1% increased") % i->first);
+            i->second->globalPriority = b->second;
+            i->second->propagatePriorities();
+        }
+        ++i;
     }
 }
 

src/hydra-queue-runner/state.hh

@@ -71,6 +71,7 @@ struct Build
     std::string projectName, jobsetName, jobName;
     time_t timestamp;
     unsigned int maxSilentTime, buildTimeout;
+    int globalPriority;
 
     std::shared_ptr<Step> toplevel;
 
@@ -80,6 +81,8 @@ struct Build
     {
         return projectName + ":" + jobsetName + ":" + jobName;
     }
+
+    void propagatePriorities();
 };
 
 
@@ -113,7 +116,11 @@ struct Step
         /* Point in time after which the step can be retried. */
         system_time after;
 
-        /* The lowest build ID depending on this step. */
+        /* The highest global priority of any build depending on this
+           step. */
+        int highestGlobalPriority{0};
+
+        /* The lowest ID of any build depending on this step. */
         BuildID lowestBuildID{std::numeric_limits<BuildID>::max()};
     };
 
@@ -282,9 +289,11 @@ private:
 
     void queueMonitorLoop();
 
+    /* Check the queue for new builds. */
     void getQueuedBuilds(Connection & conn, std::shared_ptr<nix::StoreAPI> store, unsigned int & lastBuildId);
 
-    void removeCancelledBuilds(Connection & conn);
+    /* Handle cancellation, deletion and priority bumps. */
+    void processQueueChange(Connection & conn);
 
     Step::ptr createStep(std::shared_ptr<nix::StoreAPI> store, const nix::Path & drvPath,
         Build::ptr referringBuild, Step::ptr referringStep, std::set<nix::Path> & finishedDrvs,

src/lib/Hydra/Controller/Build.pm

@@ -475,6 +475,23 @@ sub keep : Chained('buildChain') PathPart Args(1) {
 }
 
 
+sub bump : Chained('buildChain') PathPart('bump') {
+    my ($self, $c, $x) = @_;
+
+    my $build = $c->stash->{build};
+
+    requireProjectOwner($c, $build->project); # FIXME: require admin?
+
+    $c->model('DB')->schema->txn_do(sub {
+        $build->update({globalpriority => time()});
+    });
+
+    $c->flash->{successMsg} = "Build has been bumped to the front of the queue.";
+
+    $c->res->redirect($c->uri_for($self->action_for("build"), $c->req->captures));
+}
+
+
 sub add_to_release : Chained('buildChain') PathPart('add-to-release') Args(0) {
     my ($self, $c) = @_;
 

src/lib/Hydra/Controller/Root.pm

@@ -88,7 +88,7 @@ sub queue_GET {
     $c->stash->{flashMsg} //= $c->flash->{buildMsg};
     $self->status_ok(
         $c,
-        entity => [$c->model('DB::Builds')->search({finished => 0}, { order_by => ["id"]})]
+        entity => [$c->model('DB::Builds')->search({finished => 0}, { order_by => ["globalpriority desc", "id"]})]
     );
 }
 

src/lib/Hydra/Schema/Builds.pm

@@ -138,6 +138,12 @@ __PACKAGE__->table("Builds");
   default_value: 0
   is_nullable: 0
 
+=head2 globalpriority
+
+  data_type: 'integer'
+  default_value: 0
+  is_nullable: 0
+
 =head2 busy
 
   data_type: 'integer'
@@ -241,6 +247,8 @@ __PACKAGE__->add_columns(
   { data_type => "text", is_nullable => 1 },
   "priority",
   { data_type => "integer", default_value => 0, is_nullable => 0 },
+  "globalpriority",
+  { data_type => "integer", default_value => 0, is_nullable => 0 },
   "busy",
   { data_type => "integer", default_value => 0, is_nullable => 0 },
   "locker",
@@ -550,8 +558,8 @@ __PACKAGE__->many_to_many(
 );
 
 
-# Created by DBIx::Class::Schema::Loader v0.07043 @ 2015-07-30 16:52:20
-# DO NOT MODIFY THIS OR ANYTHING ABOVE! md5sum:Y2lDtgY8EBLOuCHAI8fWRQ
+# Created by DBIx::Class::Schema::Loader v0.07043 @ 2015-08-10 15:10:41
+# DO NOT MODIFY THIS OR ANYTHING ABOVE! md5sum:rjifgnPtjY96MaQ7eiGzaA
 
 __PACKAGE__->has_many(
   "dependents",

src/root/build.tt

@@ -96,6 +96,7 @@
           <li><a href="[% c.uri_for('/build' build.id 'restart') %]">Restart</a></li>
         [% ELSE %]
           <li><a href="[% c.uri_for('/build' build.id 'cancel') %]">Cancel</a></li>
+          <li><a href="[% c.uri_for('/build' build.id 'bump') %]">Bump up</a></li>
         [% END %]
         [% IF available && project.releases %]
           [% INCLUDE menuItem

src/sql/hydra.sql

@@ -156,9 +156,13 @@ create table Builds (
     nixExprInput  text,
     nixExprPath   text,
 
-    -- Information about scheduled builds.
+    -- Priority within a jobset, set via meta.schedulingPriority.
     priority      integer not null default 0,
 
+    -- Priority among all builds, used by the admin to bump builds to
+    -- the front of the queue via the web interface.
+    globalPriority integer not null default 0,
+
     -- FIXME: remove (obsolete with the new queue runner)
     busy          integer not null default 0, -- true means someone is building this job now
     locker        text, -- !!! hostname/pid of the process building this job?
@@ -218,6 +222,10 @@ create function notifyBuildCancelled() returns trigger as 'begin notify builds_c
 create trigger BuildCancelled after update on Builds for each row
   when (old.finished = 0 and new.finished = 1 and new.buildStatus = 4) execute procedure notifyBuildCancelled();
 
+create function notifyBuildBumped() returns trigger as 'begin notify builds_bumped; return null; end;' language plpgsql;
+create trigger BuildBumped after update on Builds for each row
+  when (old.globalPriority != new.globalPriority) execute procedure notifyBuildBumped();
+
 #endif
 
 

src/sql/upgrade-40.sql

@@ -0,0 +1,5 @@
+alter table Builds add column globalPriority integer not null default 0;
+
+create function notifyBuildBumped() returns trigger as 'begin notify builds_bumped; return null; end;' language plpgsql;
+create trigger BuildBumped after update on Builds for each row
+  when (old.globalPriority != new.globalPriority) execute procedure notifyBuildBumped();