fe perf: insertAdjacentHTML is faster than innerHTML+=

as a further enhancement to efafda16f, schedule a render immediately if
500ms have passed without one, and cap the max delay to 500ms.
This removes delay for short logs, and prevents an issue where output that
is produced reasonably quickly will not be rendered for a long time.

UserManual.md

@@ -19,7 +19,7 @@ Throughout this document, the fixed base path `/var/lib/laminar` is used. This i
 
 Since Debian Bullseye, Laminar is available in [the official repositories](https://packages.debian.org/search?searchon=sourcenames&keywords=laminar).
 
-Alternatively, pre-built upstream packages are available for Debian 10 (Bullseye) on x86_64 and armhf, and for Rocky/CentOS/RHEL 7 and 8 on x86_64.
+Alternatively, pre-built upstream packages are available for Debian 10 (Buster) on x86_64 and armhf, and for Rocky/CentOS/RHEL 7 and 8 on x86_64.
 
 Finally, Laminar may be built from source for any Linux distribution.
 
@@ -222,13 +222,13 @@ Then, point `laminarc` to the new location using an environment variable:
 LAMINAR_HOST=192.168.1.1:9997 laminarc queue example
 ```
 
-If you need more flexibility, consider running the communication channel as a regular unix socket and applying user and group permissions to the file. To achieve this, set
+If you need more flexibility, consider running the communication channel as a regular unix socket. Setting
 
 ```
 LAMINAR_BIND_RPC=unix:/var/run/laminar.sock
 ```
 
-or similar path in `/etc/laminar.conf`.
+or similar path in `/etc/laminar.conf` will result in a socket with group read/write permissions (`660`), so any user in the `laminar` group can queue a job.
 
 This can be securely and flexibly combined with remote triggering using `ssh`. There is no need to allow the client full shell access to the server machine, the ssh server can restrict certain users to certain commands (in this case `laminarc`). See [the authorized_keys section of the sshd man page](https://man.openbsd.org/sshd#AUTHORIZED_KEYS_FILE_FORMAT) for further information.
 
@@ -713,6 +713,7 @@ Finally, variables supplied on the command-line call to `laminarc queue`, `lamin
 - `queue [JOB [PARAMS...]]...` adds one or more jobs to the queue with optional parameters, returning immediately.
 - `start [JOB [PARAMS...]]...` starts one or more jobs with optional parameters, returning when the jobs begin execution.
 - `run [JOB [PARAMS...]]...` triggers one or more jobs with optional parameters and waits for the completion of all jobs.
+- `--next` may be passed before `JOB` in order to place the job at the front of the queue instead of at the end.
 - `set [VARIABLE=VALUE]...` sets one or more variables to be exported in subsequent scripts for the run identified by the `$JOB` and `$RUN` environment variables
 - `show-jobs` shows the known jobs on the server (`$LAMINAR_HOME/cfg/jobs/*.run`).
 - `show-running` shows the currently running jobs with their numbers.

etc/laminarc.1

@@ -6,8 +6,8 @@
 Laminar CI client application
 .Sh SYNOPSIS
 .Nm laminarc Li queue \fIJOB\fR [\fIPARAM=VALUE...\fR] ...
-.Nm laminarc Li queue \fIJOB\fR [\fIPARAM=VALUE...\fR] ...
-.Nm laminarc Li queue \fIJOB\fR [\fIPARAM=VALUE...\fR] ...
+.Nm laminarc Li start \fIJOB\fR [\fIPARAM=VALUE...\fR] ...
+.Nm laminarc Li run \fIJOB\fR [\fIPARAM=VALUE...\fR] ...
 .Nm laminarc Li set \fIPARAM=VALUE...\fR
 .Nm laminarc Li show-jobs
 .Nm laminarc Li show-running
@@ -27,6 +27,9 @@ begin execution.
 adds job(s) (with optional parameters) to the queue and returns when the jobs
 complete execution. The exit code will be non-zero if any of the runs does
 not complete successfully.
+.It \t
+\fB--next\fR may be passed to \fBqueue\fR, \fBstart\fR or \fBrun\fR in order
+to place the job at the front of the queue instead of at the end.
 .It Sy set
 sets one or more parameters to be exported as environment variables in subsequent
 scripts for the run identified by the $JOB and $RUN environment variables.

examples/nginx-ssl-reverse-proxy.conf

@@ -38,7 +38,7 @@ server {
     # fine-grained control of permissions.
     # see http://nginx.org/en/docs/http/ngx_http_proxy_module.html#proxy_pass
     # and https://laminar.ohwg.net/docs.html#Running-on-a-different-HTTP-port-or-Unix-socket
-    proxy_pass http://127.0.0.1:8080;
+    proxy_pass http://127.0.0.1:8080/;
 
     # required to allow laminar's SSE stream to pass correctly
     proxy_http_version 1.1;

src/client.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015-2020 Oliver Giles
+/// Copyright 2015-2022 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -87,12 +87,14 @@ static void printTriggerLink(const char* job, uint run) {
 
 static void usage(std::ostream& out) {
     out << "laminarc version " << laminar_version() << "\n";
-    out << "Usage: laminarc [-h|--help] COMMAND [PARAMETERS...]]\n";
+    out << "Usage: laminarc [-h|--help] COMMAND\n";
     out << "  -h|--help       show this help message\n";
     out << "where COMMAND is:\n";
     out << "  queue JOB_LIST...     queues one or more jobs for execution and returns immediately.\n";
     out << "  start JOB_LIST...     queues one or more jobs for execution and blocks until it starts.\n";
     out << "  run JOB_LIST...       queues one or more jobs for execution and blocks until it finishes.\n";
+    out << "                        JOB_LIST may be prepended with --next, in this case the job will\n";
+    out << "                        be pushed to the front of the queue instead of the end.\n";
     out << "  set PARAMETER_LIST... sets the given parameters as environment variables in the currently\n";
     out << "                        running job. Fails if run outside of a job context.\n";
     out << "  abort NAME NUMBER     aborts the run identified by NAME and NUMBER.\n";
@@ -132,16 +134,25 @@ int main(int argc, char** argv) {
 
     auto& waitScope = client.getWaitScope();
 
-    if(strcmp(argv[1], "queue") == 0) {
-        if(argc < 3) {
-            fprintf(stderr, "Usage %s queue <jobName>\n", argv[0]);
+    int jobNameIndex = 2;
+    bool frontOfQueue = false;
+
+    if(strcmp(argv[1], "queue") == 0 || strcmp(argv[1], "start") == 0 || strcmp(argv[1], "run") == 0) {
+        if(argc < 3 || (strcmp(argv[2], "--next") == 0 && argc < 4)) {
+            fprintf(stderr, "Usage %s %s JOB_LIST...\n", argv[0], argv[1]);
             return EXIT_BAD_ARGUMENT;
         }
-        int jobNameIndex = 2;
-        // make a request for each job specified on the commandline
+        if(strcmp(argv[2], "--next") == 0) {
+            frontOfQueue = true;
+            jobNameIndex++;
+        }
+    }
+
+    if(strcmp(argv[1], "queue") == 0) {
         do {
             auto req = laminar.queueRequest();
             req.setJobName(argv[jobNameIndex]);
+            req.setFrontOfQueue(frontOfQueue);
             int n = setParams(argc - jobNameIndex - 1, &argv[jobNameIndex + 1], req);
             ts.add(req.send().then([&ret,argv,jobNameIndex](capnp::Response<LaminarCi::QueueResults> resp){
                 if(resp.getResult() != LaminarCi::MethodResult::SUCCESS) {
@@ -153,16 +164,10 @@ int main(int argc, char** argv) {
             jobNameIndex += n + 1;
         } while(jobNameIndex < argc);
     } else if(strcmp(argv[1], "start") == 0) {
-        if(argc < 3) {
-            fprintf(stderr, "Usage %s queue <jobName>\n", argv[0]);
-            return EXIT_BAD_ARGUMENT;
-        }
-        kj::Vector<capnp::RemotePromise<LaminarCi::StartResults>> promises;
-        int jobNameIndex = 2;
-        // make a request for each job specified on the commandline
         do {
             auto req = laminar.startRequest();
             req.setJobName(argv[jobNameIndex]);
+            req.setFrontOfQueue(frontOfQueue);
             int n = setParams(argc - jobNameIndex - 1, &argv[jobNameIndex + 1], req);
             ts.add(req.send().then([&ret,argv,jobNameIndex](capnp::Response<LaminarCi::StartResults> resp){
                 if(resp.getResult() != LaminarCi::MethodResult::SUCCESS) {
@@ -174,21 +179,16 @@ int main(int argc, char** argv) {
             jobNameIndex += n + 1;
         } while(jobNameIndex < argc);
     } else if(strcmp(argv[1], "run") == 0) {
-        if(argc < 3) {
-            fprintf(stderr, "Usage %s run <jobName>\n", argv[0]);
-            return EXIT_BAD_ARGUMENT;
-        }
-        int jobNameIndex = 2;
-        // make a request for each job specified on the commandline
         do {
             auto req = laminar.runRequest();
             req.setJobName(argv[jobNameIndex]);
+            req.setFrontOfQueue(frontOfQueue);
             int n = setParams(argc - jobNameIndex - 1, &argv[jobNameIndex + 1], req);
             ts.add(req.send().then([&ret,argv,jobNameIndex](capnp::Response<LaminarCi::RunResults> resp){
                 if(resp.getResult() == LaminarCi::JobResult::UNKNOWN)
                     fprintf(stderr, "Failed to start job '%s'\n", argv[2]);
-		else
-		    printTriggerLink(argv[jobNameIndex], resp.getBuildNum());
+                else
+                    printTriggerLink(argv[jobNameIndex], resp.getBuildNum());
                 if(resp.getResult() != LaminarCi::JobResult::SUCCESS)
                     ret = EXIT_RUN_FAILED;
             }));
@@ -233,7 +233,7 @@ int main(int argc, char** argv) {
         }
         auto queued = laminar.listQueuedRequest().send().wait(waitScope);
         for(auto it : queued.getResult()) {
-            printf("%s\n", it.cStr());
+            printf("%s:%d\n", it.getJob().cStr(), it.getBuildNum());
         }
     } else if(strcmp(argv[1], "show-running") == 0) {
         if(argc != 2) {

src/laminar.capnp

@@ -2,10 +2,10 @@
 
 interface LaminarCi {
 
-    queue @0 (jobName :Text, params :List(JobParam)) -> (result :MethodResult, buildNum :UInt32);
-    start @1 (jobName :Text, params :List(JobParam)) -> (result :MethodResult, buildNum :UInt32);
-    run @2 (jobName :Text, params :List(JobParam)) -> (result :JobResult, buildNum :UInt32);
-    listQueued @3 () -> (result :List(Text));
+    queue @0 (jobName :Text, params :List(JobParam), frontOfQueue :Bool) -> (result :MethodResult, buildNum :UInt32);
+    start @1 (jobName :Text, params :List(JobParam), frontOfQueue :Bool) -> (result :MethodResult, buildNum :UInt32);
+    run @2 (jobName :Text, params :List(JobParam), frontOfQueue :Bool) -> (result :JobResult, buildNum :UInt32);
+    listQueued @3 () -> (result :List(Run));
     listRunning @4 () -> (result :List(Run));
     listKnown @5 () -> (result :List(Text));
     abort @6 (run :Run) -> (result :MethodResult);

src/laminar.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015-2020 Oliver Giles
+/// Copyright 2015-2022 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -306,13 +306,17 @@ std::string Laminar::getStatus(MonitorScope scope) {
             j.EndObject();
         }
         j.EndArray();
-        int nQueued = 0;
+        j.startArray("queued");
         for(const auto& run : queuedJobs) {
             if (run->name == scope.job) {
-                nQueued++;
+                j.StartObject();
+                j.set("number", run->build);
+                j.set("result", to_string(RunState::QUEUED));
+                j.set("reason", run->reason());
+                j.EndObject();
             }
         }
-        j.set("nQueued", nQueued);
+        j.EndArray();
         db->stmt("SELECT number,startedAt FROM builds WHERE name = ? AND result = ? "
                  "ORDER BY completedAt DESC LIMIT 1")
         .bind(scope.job, int(RunState::SUCCESS))
@@ -399,6 +403,8 @@ std::string Laminar::getStatus(MonitorScope scope) {
         for(const auto& run : queuedJobs) {
             j.StartObject();
             j.set("name", run->name);
+            j.set("number", run->build);
+            j.set("result", to_string(RunState::QUEUED));
             j.EndObject();
         }
         j.EndArray();
@@ -579,7 +585,7 @@ bool Laminar::loadConfiguration() {
     return true;
 }
 
-std::shared_ptr<Run> Laminar::queueJob(std::string name, ParamMap params) {
+std::shared_ptr<Run> Laminar::queueJob(std::string name, ParamMap params, bool frontOfQueue) {
     if(!fsHome->exists(kj::Path{"cfg","jobs",name+".run"})) {
         LLOG(ERROR, "Non-existent job", name);
         return nullptr;
@@ -590,7 +596,10 @@ std::shared_ptr<Run> Laminar::queueJob(std::string name, ParamMap params) {
         jobContexts.at(name).insert("default");
 
     std::shared_ptr<Run> run = std::make_shared<Run>(name, ++buildNums[name], kj::mv(params), homePath.clone());
-    queuedJobs.push_back(run);
+    if(frontOfQueue)
+        queuedJobs.push_front(run);
+    else
+        queuedJobs.push_back(run);
 
     db->stmt("INSERT INTO builds(name,number,queuedAt,parentJob,parentBuild,reason) VALUES(?,?,?,?,?,?)")
      .bind(run->name, run->build, run->queuedAt, run->parentName, run->parentBuild, run->reason())
@@ -602,6 +611,9 @@ std::shared_ptr<Run> Laminar::queueJob(std::string name, ParamMap params) {
         .startObject("data")
         .set("name", name)
         .set("number", run->build)
+        .set("result", to_string(RunState::QUEUED))
+        .set("queueIndex", frontOfQueue ? 0 : (queuedJobs.size() - 1))
+        .set("reason", run->reason())
         .EndObject();
     http->notifyEvent(j.str(), name.c_str());
 

src/laminar.h

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015-2020 Oliver Giles
+/// Copyright 2015-2022 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -52,7 +52,7 @@ public:
 
     // Queues a job, returns immediately. Return value will be nullptr if
     // the supplied name is not a known job.
-    std::shared_ptr<Run> queueJob(std::string name, ParamMap params = ParamMap());
+    std::shared_ptr<Run> queueJob(std::string name, ParamMap params = ParamMap(), bool frontOfQueue = false);
 
     // Return the latest known number of the named job
     uint latestRun(std::string job);

src/resources/index.html

@@ -22,7 +22,12 @@
   <nav>
    <table class="table striped">
     <tr v-for="job in jobsQueued">
-     <td><router-link  :to="'jobs/'+job.name">{{job.name}}</router-link> <i>queued</i></td>
+     <td>
+      <span v-html="runIcon(job.result)"></span>
+      <router-link :to="'jobs/'+job.name">{{job.name}}</router-link>
+      <router-link :to="'jobs/'+job.name+'/'+job.number">#{{job.number}}</router-link>
+      <i>queued</i>
+     </td>
     </tr>
     <tr v-for="job in jobsRunning">
      <td>
@@ -134,15 +139,11 @@
      <th class="text-center">Duration <a class="sort" :class="(sort.field=='duration'?sort.order:'')" v-on:click="do_sort('duration')">&nbsp;</a></th>
      <th class="text-center vp-sm-hide">Reason <a class="sort" :class="(sort.field=='reason'?sort.order:'')" v-on:click="do_sort('reason')">&nbsp;</a></th>
     </tr></thead>
-    <tr v-show="nQueued">
-     <td style="width:1px"><span v-html="runIcon('queued')"></span></td>
-     <td colspan="4"><i>{{nQueued}} run(s) queued</i></td>
-    </tr>
-    <tr v-for="job in jobsRunning.concat(jobsRecent)" track-by="$index">
+    <tr v-for="job in jobsQueued.concat(jobsRunning).concat(jobsRecent)" track-by="$index">
      <td style="width:1px"><span v-html="runIcon(job.result)"></span></td>
      <td><router-link :to="'jobs/'+route.params.name+'/'+job.number">#{{job.number}}</router-link></td>
-     <td class="text-center">{{formatDate(job.started)}}</td>
-     <td class="text-center">{{formatDuration(job.started, job.completed)}}</td>
+     <td class="text-center"><span v-if="job.result!='queued'">{{formatDate(job.started)}}</span></td>
+     <td class="text-center"><span v-if="job.result!='queued'">{{formatDuration(job.started, job.completed)}}</span></td>
      <td class="text-center vp-sm-hide">{{job.reason}}</td>
     </tr>
    </table>
@@ -186,14 +187,14 @@
    </div>
   </div>
   <div class="console-log">
-   <code v-html="log"></code>
-   <span v-show="job.result == 'running'" v-html="runIcon('running')" style="display: block;"></span>
+   <code></code>
+   <span v-show="!logComplete" v-html="runIcon('running')" style="display: block;"></span>
   </div>
  </div></template>
 
  <main id="app" style="display: grid; grid-template-rows: auto 1fr auto; height: 100%;">
   <nav id="nav-top" style="display: grid; grid-template-columns: auto auto 1fr auto auto; grid-gap: 15px;">
-   <router-link to="/" style="display: grid; grid-auto-flow: column; align-items: center; margin: 5px; font-size: 20px;">
+   <router-link to="." style="display: grid; grid-auto-flow: column; align-items: center; margin: 5px; font-size: 20px;">
     <img src="icon.png"> {{title}}
    </router-link>
    <div id="nav-top-links" style="display: grid; grid-auto-flow: column; justify-content: start; gap: 15px; padding: 0 15px; align-items: center; font-size: 16px;">

src/resources/js/app.js

@@ -397,8 +397,8 @@ const Home = templateId => {
     data: () => state,
     methods: {
       status: function(msg) {
-        state.jobsQueued = msg.queued;
-        state.jobsRunning = msg.running;
+        state.jobsQueued = msg.queued.reverse();
+        state.jobsRunning = msg.running.reverse();
         state.jobsRecent = msg.recent;
         state.resultChanged = msg.resultChanged;
         state.lowPassRates = msg.lowPassRates;
@@ -415,7 +415,7 @@ const Home = templateId => {
         });
       },
       job_queued: function(data) {
-        state.jobsQueued.splice(0, 0, data);
+        state.jobsQueued.splice(state.jobsQueued.length - data.queueIndex, 0, data);
         this.$forceUpdate();
       },
       job_started: function(data) {
@@ -559,11 +559,11 @@ const All = templateId => {
 const Job = templateId => {
   const state = {
     description: '',
+    jobsQueued: [],
     jobsRunning: [],
     jobsRecent: [],
     lastSuccess: null,
     lastFailed: null,
-    nQueued: 0,
     pages: 0,
     sort: {}
   };
@@ -575,11 +575,11 @@ const Job = templateId => {
     methods: {
       status: function(msg) {
         state.description = msg.description;
-        state.jobsRunning = msg.running;
+        state.jobsQueued = msg.queued.reverse();
+        state.jobsRunning = msg.running.reverse();
         state.jobsRecent = msg.recent;
         state.lastSuccess = msg.lastSuccess;
         state.lastFailed = msg.lastFailed;
-        state.nQueued = msg.nQueued;
         state.pages = msg.pages;
         state.sort = msg.sort;
 
@@ -593,11 +593,12 @@ const Job = templateId => {
           chtBuildTime = Charts.createRunTimeChart("chartBt", msg.recent, msg.averageRuntime);
         });
       },
-      job_queued: function() {
-        state.nQueued++;
+      job_queued: function(data) {
+        state.jobsQueued.splice(state.jobsQueued.length - data.queueIndex, 0, data);
+        this.$forceUpdate();
       },
       job_started: function(data) {
-        state.nQueued--;
+        state.jobsQueued.splice(state.jobsQueued.length - data.queueIndex - 1, 1);
         state.jobsRunning.splice(0, 0, data);
         this.$forceUpdate();
       },
@@ -642,7 +643,7 @@ const Run = templateId => {
   const state = {
     job: { artifacts: [], upstream: {} },
     latestNum: null,
-    log: '',
+    logComplete: false,
   };
   const logFetcher = (vm, name, num) => {
     const abort = new AbortController();
@@ -650,18 +651,62 @@ const Run = templateId => {
       // ATOW pipeThrough not supported in Firefox
       //const reader = res.body.pipeThrough(new TextDecoderStream).getReader();
       const reader = res.body.getReader();
+      const target = document.getElementsByTagName('code')[0];
+      let logToRender = '';
+      let logComplete = false;
+      let tid = null;
+      let lastUiUpdate = 0;
+
+      function updateUI() {
+        // output may contain private ANSI CSI escape sequence to point to
+        // downstream jobs. ansi_up (correctly) discards unknown sequences,
+        // so they must be matched before passing through ansi_up. ansi_up
+        // also (correctly) escapes HTML, so they need to be converted back
+        // to links after going through ansi_up.
+        // A better solution one day would be if ansi_up were to provide
+        // a callback interface for handling unknown sequences.
+        // Also, update the DOM directly rather than using a binding through
+        // Vue, the performance is noticeably better with large logs.
+        target.insertAdjacentHTML('beforeend', ansi_up.ansi_to_html(
+          logToRender.replace(/\033\[\{([^:]+):(\d+)\033\\/g, (m, $1, $2) =>
+            '~~~~LAMINAR_RUN~'+$1+':'+$2+'~'
+          )
+        ).replace(/~~~~LAMINAR_RUN~([^:]+):(\d+)~/g, (m, $1, $2) =>
+          '<a href="jobs/'+$1+'" onclick="return LaminarApp.navigate(this.href);">'+$1+'</a>:'+
+          '<a href="jobs/'+$1+'/'+$2+'" onclick="return LaminarApp.navigate(this.href);">#'+$2+'</a>'
+        ));
+        logToRender = '';
+        if (logComplete) {
+          // output finished
+          state.logComplete = true;
+        }
+
+        lastUiUpdate = Date.now();
+        tid = null;
+      }
+
       return function pump() {
         return reader.read().then(({done, value}) => {
-          value = utf8decoder.decode(value);
-          if (done)
+          if (done) {
+            // do not set state.logComplete directly, because rendering
+            // may take some time, and we don't want the progress indicator
+            // to disappear before rendering is complete. Instead, delay
+            // it until after the entire log has been rendered
+            logComplete = true;
+            // if no render update is pending, schedule one immediately
+            // (do not use the delayed buffering mechanism from below), so
+            // that for the common case of short logs, the loading spinner
+            // disappears immediately as the log is rendered
+            if(tid === null)
+              setTimeout(updateUI, 0);
             return;
-          state.log += ansi_up.ansi_to_html(
-            value.replace(/\033\[\{([^:]+):(\d+)\033\\/g, (m, $1, $2) =>
-                   '<a href="jobs/'+$1+'" onclick="return vroute(this);">'+$1+'</a>:'+
-                   '<a href="jobs/'+$1+'/'+$2+'" onclick="return vroute(this);">#'+$2+'</a>'
-                 )
-          );
-          vm.$forceUpdate();
+          }
+          // sometimes logs can be very large, and we are calling pump()
+          // furiously to get all the data to the client. To prevent straining
+          // the client renderer, buffer the data and delay the UI updates.
+          logToRender += utf8decoder.decode(value);
+          if(tid === null)
+            tid = setTimeout(updateUI, Math.max(500 - (Date.now() - lastUiUpdate), 0));
           return pump();
         });
       }();
@@ -684,7 +729,9 @@ const Run = templateId => {
         state.job = data;
         state.latestNum = data.latestNum;
         state.jobsRunning = [data];
-        state.log = '';
+        state.logComplete = false;
+        // DOM is used directly for performance
+        document.getElementsByTagName('code')[0].innerHTML = '';
         if(this.logstream)
           this.logstream.abort();
         if(data.started)
@@ -825,7 +872,7 @@ Vue.component('RouterView', (() => {
   };
 })());
 
-new Vue({
+const LaminarApp = new Vue({
   el: '#app',
   data: {
     title: '', // populated by status message
@@ -851,6 +898,11 @@ new Vue({
         new Notification('Job ' + data.result, {
           body: data.name + ' ' + '#' + data.number + ': ' + data.result
         });
+    },
+    navigate: function(path) {
+      history.pushState(null, null, path);
+      this.$emit('navigate');
+      return false;
     }
   },
   watch: {

src/rpc.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015-2019 Oliver Giles
+/// Copyright 2015-2022 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -53,7 +53,7 @@ public:
     kj::Promise<void> queue(QueueContext context) override {
         std::string jobName = context.getParams().getJobName();
         LLOG(INFO, "RPC queue", jobName);
-        std::shared_ptr<Run> run = laminar.queueJob(jobName, params(context.getParams().getParams()));
+        std::shared_ptr<Run> run = laminar.queueJob(jobName, params(context.getParams().getParams()), context.getParams().getFrontOfQueue());
         if(Run* r = run.get()) {
             context.getResults().setResult(LaminarCi::MethodResult::SUCCESS);
             context.getResults().setBuildNum(r->build);
@@ -67,7 +67,7 @@ public:
     kj::Promise<void> start(StartContext context) override {
         std::string jobName = context.getParams().getJobName();
         LLOG(INFO, "RPC start", jobName);
-        std::shared_ptr<Run> run = laminar.queueJob(jobName, params(context.getParams().getParams()));
+        std::shared_ptr<Run> run = laminar.queueJob(jobName, params(context.getParams().getParams()), context.getParams().getFrontOfQueue());
         if(Run* r = run.get()) {
             return r->whenStarted().then([context,r]() mutable {
                 context.getResults().setResult(LaminarCi::MethodResult::SUCCESS);
@@ -83,7 +83,7 @@ public:
     kj::Promise<void> run(RunContext context) override {
         std::string jobName = context.getParams().getJobName();
         LLOG(INFO, "RPC run", jobName);
-        std::shared_ptr<Run> run = laminar.queueJob(jobName, params(context.getParams().getParams()));
+        std::shared_ptr<Run> run = laminar.queueJob(jobName, params(context.getParams().getParams()), context.getParams().getFrontOfQueue());
         if(run) {
             return run->whenFinished().then([context,run](RunState state) mutable {
                 context.getResults().setResult(fromRunState(state));
@@ -101,7 +101,9 @@ public:
         auto res = context.getResults().initResult(queue.size());
         int i = 0;
         for(auto it : queue) {
-            res.set(i++, it->name);
+            res[i].setJob(it->name);
+            res[i].setBuildNum(it->build);
+            i++;
         }
         return kj::READY_NOW;
     }

src/server.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015-2019 Oliver Giles
+/// Copyright 2015-2021 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -30,6 +30,7 @@
 #include <sys/eventfd.h>
 #include <sys/inotify.h>
 #include <sys/signalfd.h>
+#include <sys/stat.h>
 
 // Size of buffer used to read from file descriptors. Should be
 // a multiple of sizeof(struct signalfd_siginfo) == 128
@@ -117,8 +118,11 @@ void Server::listenRpc(Rpc &rpc, kj::StringPtr rpcBindAddress)
     if(rpcBindAddress.startsWith("unix:"))
         unlink(rpcBindAddress.slice(strlen("unix:")).cStr());
     listeners->add(ioContext.provider->getNetwork().parseAddress(rpcBindAddress)
-              .then([this,&rpc](kj::Own<kj::NetworkAddress>&& addr) {
-        return acceptRpcClient(rpc, addr->listen());
+              .then([this,&rpc,rpcBindAddress](kj::Own<kj::NetworkAddress>&& addr) {
+        kj::Own<kj::ConnectionReceiver> listener = addr->listen();
+        if(rpcBindAddress.startsWith("unix:"))
+            chmod(rpcBindAddress.slice(strlen("unix:")).cStr(), 0660);
+        return acceptRpcClient(rpc, kj::mv(listener));
     }));
 
 }
@@ -128,8 +132,19 @@ void Server::listenHttp(Http &http, kj::StringPtr httpBindAddress)
     if(httpBindAddress.startsWith("unix:"))
         unlink(httpBindAddress.slice(strlen("unix:")).cStr());
     listeners->add(ioContext.provider->getNetwork().parseAddress(httpBindAddress)
-              .then([this,&http](kj::Own<kj::NetworkAddress>&& addr) {
-        return http.startServer(ioContext.lowLevelProvider->getTimer(), addr->listen());
+              .then([this,&http,httpBindAddress](kj::Own<kj::NetworkAddress>&& addr) {
+        kj::Own<kj::ConnectionReceiver> listener = addr->listen();
+        if(httpBindAddress.startsWith("unix:"))
+            chmod(httpBindAddress.slice(strlen("unix:")).cStr(), 0660);
+        return http.startServer(ioContext.lowLevelProvider->getTimer(), kj::mv(listener));
+    }).catch_([this,&http,httpBindAddress](kj::Exception&&e) mutable -> kj::Promise<void> {
+        if(e.getType() == kj::Exception::Type::DISCONNECTED) {
+            LLOG(ERROR, "HTTP disconnect, restarting server", e.getDescription());
+            listenHttp(http, httpBindAddress);
+            return kj::READY_NOW;
+        }
+        // otherwise propagate the exception
+        return kj::mv(e);
     }));
 }
 
@@ -157,5 +172,7 @@ kj::Promise<void> Server::handleFdRead(kj::AsyncInputStream* stream, char* buffe
 
 void Server::taskFailed(kj::Exception &&exception) {
     //kj::throwFatalException(kj::mv(exception));
-    fprintf(stderr, "taskFailed: %s\n", exception.getDescription().cStr());
+    // prettier
+    fprintf(stderr, "fatal: %s\n", exception.getDescription().cStr());
+    exit(EXIT_FAILURE);
 }

test/laminar-fixture.h

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2019-2020 Oliver Giles
+/// Copyright 2019-2022 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -33,20 +33,26 @@
 class LaminarFixture : public ::testing::Test {
 public:
     LaminarFixture() {
-        bind_rpc = std::string("unix:/") + tmp.path.toString(true).cStr() + "/rpc.sock";
-        bind_http = std::string("unix:/") + tmp.path.toString(true).cStr() + "/http.sock";
         home = tmp.path.toString(true).cStr();
-        tmp.fs->openSubdir(kj::Path{"cfg", "jobs"}, kj::WriteMode::CREATE | kj::WriteMode::CREATE_PARENT);
+        bind_rpc = std::string("unix:/") + home + "/rpc.sock";
+        bind_http = std::string("unix:/") + home + "/http.sock";
         settings.home = home.c_str();
         settings.bind_rpc = bind_rpc.c_str();
         settings.bind_http = bind_http.c_str();
         settings.archive_url = "/test-archive/";
+    }
+    ~LaminarFixture() noexcept(true) {}
+
+    void SetUp() override {
+        tmp.init();
         server = new Server(*ioContext);
         laminar = new Laminar(*server, settings);
     }
-    ~LaminarFixture() noexcept(true) {
+
+    void TearDown() override {
         delete server;
         delete laminar;
+        tmp.clean();
     }
 
     kj::Own<EventSource> eventSource(const char* path) {
@@ -92,6 +98,14 @@ public:
         return { res.getResult(), kj::mv(log) };
     }
 
+    void setNumExecutors(int nexec) {
+        KJ_IF_MAYBE(f, tmp.fs->tryOpenFile(kj::Path{"cfg", "contexts", "default.conf"},
+                kj::WriteMode::CREATE | kj::WriteMode::MODIFY | kj::WriteMode::CREATE_PARENT)) {
+            std::string content = "EXECUTORS=" + std::to_string(nexec);
+            (*f)->writeAll(content);
+        }
+    }
+
     kj::String stripLaminarLogLines(const kj::String& str) {
         auto out = kj::heapString(str.size());
         char *o = out.begin();

test/laminar-functional.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2019 Oliver Giles
+/// Copyright 2019-2022 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -162,3 +162,27 @@ TEST_F(LaminarFixture, JobDescription) {
     ASSERT_TRUE(data.HasMember("description"));
     EXPECT_STREQ("bar", data["description"].GetString());
 }
+
+TEST_F(LaminarFixture, QueueFront) {
+    setNumExecutors(0);
+    defineJob("foo", "true");
+    defineJob("bar", "true");
+    auto es = eventSource("/");
+    auto req1 = client().queueRequest();
+    req1.setJobName("foo");
+    auto res1 = req1.send();
+    auto req2 = client().queueRequest();
+    req2.setFrontOfQueue(true);
+    req2.setJobName("bar");
+    auto res2 = req2.send();
+    ioContext->waitScope.poll();
+    setNumExecutors(2);
+    ioContext->waitScope.poll();
+    ASSERT_EQ(5, es->messages().size());
+    auto started1 = es->messages().at(3).GetObject();
+    EXPECT_STREQ("job_started", started1["type"].GetString());
+    EXPECT_STREQ("bar", started1["data"]["name"].GetString());
+    auto started2 = es->messages().at(4).GetObject();
+    EXPECT_STREQ("job_started", started2["type"].GetString());
+    EXPECT_STREQ("foo", started2["data"]["name"].GetString());
+}

test/tempdir.h

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2018-2020 Oliver Giles
+/// Copyright 2018-2022 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -28,12 +28,25 @@ class TempDir {
 public:
     TempDir() :
         path(mkdtemp()),
-        fs(kj::newDiskFilesystem()->getRoot().openSubdir(path, kj::WriteMode::CREATE|kj::WriteMode::MODIFY))
+        fs(kj::newDiskFilesystem()->getRoot().openSubdir(path, kj::WriteMode::MODIFY))
     {
     }
     ~TempDir() noexcept {
         kj::newDiskFilesystem()->getRoot().remove(path);
     }
+    void init() {
+        // set up empty directory structure
+        fs->openSubdir(kj::Path{"cfg"}, kj::WriteMode::CREATE);
+        fs->openSubdir(kj::Path{"cfg", "jobs"}, kj::WriteMode::CREATE);
+        fs->openSubdir(kj::Path{"cfg", "contexts"}, kj::WriteMode::CREATE);
+    }
+    void clean() {
+        // rm -rf in config folder
+        for(kj::StringPtr name : fs->listNames()) {
+            fs->remove(kj::Path{name});
+        }
+    }
+
     kj::Path path;
     kj::Own<const kj::Directory> fs;
 private: