v0.3

CMakeLists.txt

@@ -1,5 +1,5 @@
 ###
-### Copyright 2015 Oliver Giles
+### Copyright 2016 Oliver Giles
 ###
 ### This file is part of Laminar
 ###
@@ -84,7 +84,7 @@ generate_compressed_bins(${CMAKE_BINARY_DIR} js/angular.min.js js/angular-route.
 add_executable(laminard src/database.cpp src/main.cpp src/server.cpp src/laminar.cpp
     src/conf.cpp src/resources.cpp src/run.cpp laminar.capnp.c++ ${COMPRESSED_BINS})
 # TODO: some alternative to boost::filesystem?
-target_link_libraries(laminard capnp-rpc capnp kj-async kj pthread boost_filesystem boost_system sqlite3)
+target_link_libraries(laminard capnp-rpc capnp kj-async kj pthread boost_filesystem boost_system sqlite3 z)
 
 ## Client
 add_executable(laminarc src/client.cpp laminar.capnp.c++)

docker-build-centos.sh

@@ -0,0 +1,80 @@
+#!/bin/bash
+
+VERSION=0.3
+
+OUTPUT_DIR=$PWD
+
+SOURCE_DIR=$(readlink -f $(dirname ${BASH_SOURCE[0]}))
+
+docker run --rm -i -v $SOURCE_DIR:/root/rpmbuild/SOURCES/laminar-$VERSION:ro -v $OUTPUT_DIR:/output centos bash -xe <<EOS
+
+yum -y install rpm-build cmake make gcc gcc-c++ wget sqlite-devel boost-devel zlib-devel
+
+mkdir /build
+cd /build
+
+wget -O capnproto.tar.gz https://github.com/sandstorm-io/capnproto/archive/v0.5.3.tar.gz
+wget -O websocketpp.tar.gz https://github.com/zaphoyd/websocketpp/archive/0.6.0.tar.gz
+wget -O rapidjson.tar.gz https://github.com/miloyip/rapidjson/archive/v1.0.2.tar.gz
+md5sum -c <<EOF
+909bd13ad6b8bc840ac78ab8f5bcb0a4  capnproto.tar.gz
+5a485884c01f881aafbf1e055d851b82  websocketpp.tar.gz
+97cc60d01282a968474c97f60714828c  rapidjson.tar.gz
+EOF
+
+tar xzf capnproto.tar.gz
+tar xzf websocketpp.tar.gz
+tar xzf rapidjson.tar.gz
+
+cd /build/capnproto-0.5.3/c++/
+cmake -DCMAKE_BUILD_TYPE=Release -DBUILD_TESTING=off .
+make -j4
+make install
+
+cd /build/websocketpp-0.6.0/
+cmake .
+make install
+
+cd /build/rapidjson-1.0.2/
+cmake -DRAPIDJSON_BUILD_EXAMPLES=off .
+make install
+
+cd
+cat <<EOF > laminar.spec
+Summary: Lightweight Continuous Integration Service
+Name: laminar
+Version: $VERSION
+Release: 1
+License: GPL
+BuildRequires: systemd-units
+Requires: boost-filesystem zlib
+
+%description
+Lightweight Continuous Integration Service
+
+%prep
+
+%build
+cmake -DCMAKE_BUILD_TYPE=Release -DCMAKE_INSTALL_PREFIX=/ %{_sourcedir}/laminar-$VERSION
+pwd
+make
+
+%install
+%make_install
+
+%files
+%{_bindir}/laminarc
+%{_bindir}/laminard
+%{_unitdir}/laminar.service
+%config(noreplace) %{_sysconfdir}/laminar.conf
+
+%post
+echo Creating laminar user with home in %{_sharedstatedir}/laminar
+useradd -r -d %{_sharedstatedir}/laminar -s %{_sbindir}/nologin laminar
+mkdir -p %{_sharedstatedir}/laminar/cfg/{jobs,nodes,scripts}
+chown -R laminar: %{_sharedstatedir}/laminar
+EOF
+
+rpmbuild -ba laminar.spec
+mv rpmbuild/RPMS/x86_64/laminar-$VERSION-1.x86_64.rpm /output/
+EOS

docker-build-debian.sh

@@ -0,0 +1,71 @@
+#!/bin/bash
+
+VERSION=0.3
+
+OUTPUT_DIR=$PWD
+
+SOURCE_DIR=$(readlink -f $(dirname ${BASH_SOURCE[0]}))
+
+docker run --rm -i -v $SOURCE_DIR:/laminar:ro -v $OUTPUT_DIR:/output debian:stable bash -xe <<EOS
+
+apt-get update
+apt-get install -y wget cmake g++ libsqlite3-dev libboost-filesystem1.55-dev zlib1g-dev
+
+mkdir /build
+cd /build
+
+wget -O capnproto.tar.gz https://github.com/sandstorm-io/capnproto/archive/v0.5.3.tar.gz
+wget -O websocketpp.tar.gz https://github.com/zaphoyd/websocketpp/archive/0.6.0.tar.gz
+wget -O rapidjson.tar.gz https://github.com/miloyip/rapidjson/archive/v1.0.2.tar.gz
+md5sum -c <<EOF
+909bd13ad6b8bc840ac78ab8f5bcb0a4  capnproto.tar.gz
+5a485884c01f881aafbf1e055d851b82  websocketpp.tar.gz
+97cc60d01282a968474c97f60714828c  rapidjson.tar.gz
+EOF
+
+tar xzf capnproto.tar.gz
+tar xzf websocketpp.tar.gz
+tar xzf rapidjson.tar.gz
+
+cd /build/capnproto-0.5.3/c++/
+cmake -DCMAKE_BUILD_TYPE=Release -DBUILD_TESTING=off .
+make -j4
+make install
+
+cd /build/websocketpp-0.6.0/
+cmake .
+make install
+
+cd /build/rapidjson-1.0.2/
+cmake -DRAPIDJSON_BUILD_EXAMPLES=off .
+make install
+
+cd /build
+cmake -DCMAKE_BUILD_TYPE=Release -DCMAKE_INSTALL_PREFIX=/ /laminar
+make -j4
+mkdir laminar
+make DESTDIR=laminar install/strip
+
+mkdir laminar/DEBIAN
+cat <<EOF > laminar/DEBIAN/control
+Package: laminar
+Version: $VERSION
+Section: 
+Priority: optional
+Architecture: amd64
+Maintainer: Oliver Giles <web ohwg net>
+Depends: libboost-filesystem1.55.0, zlib1g
+Description: Lightweight Continuous Integration Service
+EOF
+cat <<EOF > laminar/DEBIAN/postinst
+#!/bin/bash
+echo Creating laminar user with home in /var/lib/laminar
+useradd -r -d /var/lib/laminar -s /usr/sbin/nologin laminar
+mkdir -p /var/lib/laminar/cfg/{jobs,nodes,scripts}
+chown -R laminar: /var/lib/laminar
+EOF
+chmod +x laminar/DEBIAN/postinst
+
+dpkg-deb --build laminar
+mv laminar.deb /output/laminar-$VERSION-1-amd64.deb
+EOS

src/client.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///

src/conf.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///

src/conf.h

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///

src/database.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -19,6 +19,7 @@
 #include "database.h"
 
 #include <sqlite3.h>
+#include <string.h>
 
 Database::Database(const char *path) {
     sqlite3_open(path, &hdl);
@@ -47,15 +48,17 @@ void Database::Statement::bindValue(int i, int e) {
 
 void Database::Statement::bindValue(int i, const char* e) {
     sqlite3_bind_text(stmt, i, e, -1, NULL);
-
 }
 
-void Database::Statement::bindValue(int i, std::string e) {
+void Database::Statement::bindValue(int i, const std::string& e) {
-    sqlite3_bind_text(stmt, i, e.c_str(), e.length(), NULL);
+    sqlite3_bind_text(stmt, i, e.data(), e.size(), NULL);
 }
 
 template<> std::string Database::Statement::fetchColumn(int col) {
-    return (char*)sqlite3_column_text(stmt, col);
+    int sz = sqlite3_column_bytes(stmt, col);
+    std::string res(sz, '\0');
+    memcpy(&res[0], sqlite3_column_text(stmt, col), sz);
+    return res;
 }
 
 template<> const char* Database::Statement::fetchColumn(int col) {

src/database.h

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -55,9 +55,12 @@ private:
         ~Statement();
 
         // Bind several parameters in a single call. They are bound
-        // by index in the order passed into this function
+        // by index in the order passed into this function. Must be
+        // passed by reference because arguments may be std::strings,
+        // which must be passed by reference because sqlite requires
+        // the bound string's lifetime to exist until sqlite3_step
         template<typename...Args>
-        Statement& bind(Args...args) {
+        Statement& bind(const Args&...args) {
             return bindRecursive<Args...>(1, args...);
         }
         // Fetch columns. Supply a callback that will be executed for
@@ -100,7 +103,7 @@ private:
         bool row();
 
         template<typename T, typename...Args>
-        Statement& bindRecursive(int i, T v, Args...args) {
+        Statement& bindRecursive(int i, const T& v, const Args&...args) {
             bindValue(i, v); // specialization must exist for T
             return bindRecursive(i + 1, args...);
         }
@@ -112,7 +115,7 @@ private:
         // Bind value specializations
         void bindValue(int i, int e);
         void bindValue(int i, const char* e);
-        void bindValue(int i, std::string e);
+        void bindValue(int i, const std::string& e);
 
         // Declaration for fetch column interface,
         // intentionally missing definition

src/laminar.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -19,10 +19,12 @@
 #include "laminar.h"
 #include "server.h"
 #include "conf.h"
+#include "log.h"
 
 #include <sys/wait.h>
+#include <sys/signalfd.h>
 #include <fstream>
-#include <kj/debug.h>
+#include <zlib.h>
 
 #include <boost/filesystem.hpp>
 namespace fs = boost::filesystem;
@@ -69,7 +71,12 @@ Laminar::Laminar() {
     // Prepare database for first use
     // TODO: error handling
     db->exec("CREATE TABLE IF NOT EXISTS builds("
-            "name TEXT, number INT UNSIGNED, node TEXT, queuedAt INT, startedAt INT, completedAt INT, result INT, output TEXT, parentJob TEXT, parentBuild INT, reason TEXT)");
+             "name TEXT, number INT UNSIGNED, node TEXT, queuedAt INT, "
+             "startedAt INT, completedAt INT, result INT, output TEXT, "
+             "outputLen INT, parentJob TEXT, parentBuild INT, reason TEXT, "
+             "PRIMARY KEY (name, number))");
+    db->exec("CREATE INDEX IF NOT EXISTS idx_completion_time ON builds("
+             "completedAt DESC)");
 
     // retrieve the last build numbers
     db->stmt("SELECT name, MAX(number) FROM builds GROUP BY name")
@@ -123,10 +130,16 @@ void Laminar::sendStatus(LaminarClient* client) {
         if(const Run* run = activeRun(client->scope.job, client->scope.num)) {
             client->sendMessage(run->log.c_str());
         } else { // it must be finished, fetch it from the database
-            db->stmt("SELECT output FROM builds WHERE name = ? AND number = ?")
+            db->stmt("SELECT output, outputLen FROM builds WHERE name = ? AND number = ?")
               .bind(client->scope.job, client->scope.num)
-              .fetch<const char*>([=](const char* log) {
+              .fetch<str,int>([=](str zipped, unsigned long sz) {
-                client->sendMessage(log);
+                std::string log(sz+1,'\0');
+                int res = ::uncompress((unsigned char*)&log[0], &sz,
+                        (unsigned char*)zipped.data(), zipped.size());
+                if(res == Z_OK)
+                    client->sendMessage(log);
+                else
+                    LLOG(ERROR, "Failed to uncompress log", res);
             });
         }
         return;
@@ -151,6 +164,7 @@ void Laminar::sendStatus(LaminarClient* client) {
             j.set("queued", run->startedAt - run->queuedAt);
             j.set("started", run->startedAt);
             j.set("reason", run->reason());
+            j.set("result", to_string(RunState::RUNNING));
             db->stmt("SELECT completedAt - startedAt FROM builds WHERE name = ? ORDER BY completedAt DESC LIMIT 1")
              .bind(run->name)
              .fetch<int>([&](int etc){
@@ -183,6 +197,7 @@ void Laminar::sendStatus(LaminarClient* client) {
             j.set("number", run->build);
             j.set("node", run->node->name);
             j.set("started", run->startedAt);
+            j.set("result", to_string(RunState::RUNNING));
             j.set("reason", run->reason());
             j.EndObject();
         }
@@ -312,6 +327,19 @@ void Laminar::run() {
 
     srv = new Server(*this, listen_rpc, listen_http);
 
+    // handle SIGCHLD
+    sigset_t mask;
+    sigemptyset(&mask);
+    sigaddset(&mask, SIGCHLD);
+    sigprocmask(SIG_BLOCK, &mask, NULL);
+    int sigchld = signalfd(-1, &mask, 0);
+    srv->addDescriptor(sigchld, [this](char* buf, size_t sz){
+        struct signalfd_siginfo* siginfo = (struct signalfd_siginfo*) buf;
+        KJ_ASSERT(siginfo->ssi_signo == SIGCHLD);
+        reapAdvance();
+        assignNewJobs();
+    });
+
     srv->start();
 }
 
@@ -397,14 +425,14 @@ bool Laminar::loadConfiguration() {
 
 std::shared_ptr<Run> Laminar::queueJob(std::string name, ParamMap params) {
     if(!fs::exists(fs::path(homeDir)/"cfg"/"jobs"/name)) {
-        KJ_LOG(ERROR, "Non-existent job", name);
+        LLOG(ERROR, "Non-existent job", name);
         return nullptr;
     }
 
     // attempt to create a workspace for this job if it doesn't exist
     if(!fs::exists(fs::path(homeDir)/"run"/name/"workspace")) {
         if(!fs::create_directories(fs::path(homeDir)/"run"/name/"workspace")) {
-            KJ_LOG(ERROR, "Could not create job workspace", name);
+            LLOG(ERROR, "Could not create job workspace", name);
             return nullptr;
         }
     }
@@ -421,7 +449,7 @@ std::shared_ptr<Run> Laminar::queueJob(std::string name, ParamMap params) {
             } else if(it->first == "=reason") {
                 run->reasonMsg = it->second;
             } else {
-                KJ_LOG(ERROR, "Unknown internal job parameter", it->first);
+                LLOG(ERROR, "Unknown internal job parameter", it->first);
             }
             it = params.erase(it);
         } else
@@ -460,26 +488,25 @@ kj::Promise<RunState> Laminar::waitForRun(const Run* run) {
 bool Laminar::stepRun(std::shared_ptr<Run> run) {
     bool complete = run->step();
     if(!complete) {
-        srv->addProcess(run->fd, [=](char* b,size_t n){
+        srv->addDescriptor(run->fd, [=](char* b,size_t n){
             std::string s(b,n);
             run->log += s;
             for(LaminarClient* c : clients) {
                 if(c->scope.wantsLog(run->name, run->build))
                     c->sendMessage(s);
             }
-        }, [this,run](){ reapAdvance();});
+        });
     }
     return complete;
 }
 
+// Reaps a zombie and steps the corresponding Run to its next state.
+// Should be called on SIGCHLD
 void Laminar::reapAdvance() {
     int ret = 0;
-    // TODO: If we pass WNOHANG here for better asynchronicity, how do
+    pid_t pid;
-    // we re-schedule a poll to wait for finished child processes?
+    while((pid = waitpid(-1, &ret, WNOHANG)) > 0) {
-    pid_t pid = waitpid(-1, &ret, 0);
+        LLOG(INFO, "Reaping", pid);
-    // TODO: handle signalled child processes
-    if(pid > 0) {
-        KJ_LOG(INFO, "Reaping", pid);
         auto it = activeJobs.get<0>().find(pid);
         std::shared_ptr<Run> run = *it;
         bool completed = true;
@@ -491,7 +518,6 @@ void Laminar::reapAdvance() {
         if(completed)
             run->complete();
     }
-    assignNewJobs();
 }
 
 bool Laminar::nodeCanQueue(const Node& node, const Run& run) const {
@@ -530,14 +556,16 @@ void Laminar::assignNewJobs() {
                 // create a working directory (different to a workspace!)
                 fs::path wd = fs::path(homeDir)/"run"/run->name/std::to_string(buildNum);
                 if(!fs::create_directory(wd)) {
-                    KJ_LOG(ERROR, "Could not create working directory", wd.string());
+                    LLOG(ERROR, "Could not create working directory", wd.string());
                     break;
                 }
                 run->wd = wd.string();
                 // create an archive directory
                 fs::path archive = fs::path(homeDir)/"archive"/run->name/std::to_string(buildNum);
-                if(!fs::create_directories(archive)) {
+                if(fs::is_directory(archive)) {
-                    KJ_LOG(ERROR, "Could not create archive directory", archive.string());
+                    LLOG(WARNING, "Archive directory already exists", archive.string());
+                } else if(!fs::create_directories(archive)) {
+                    LLOG(ERROR, "Could not create archive directory", archive.string());
                     break;
                 }
 
@@ -587,7 +615,7 @@ void Laminar::assignNewJobs() {
                 // update next build number
                 buildNums[run->name] = buildNum;
 
-                KJ_LOG(INFO, "Queued job to node", run->name, run->build, node.name);
+                LLOG(INFO, "Queued job to node", run->name, run->build, node.name);
 
                 // notify clients
                 Json j;
@@ -616,12 +644,12 @@ void Laminar::assignNewJobs() {
                 }
 
                 // setup run completion handler
-                run->notifyCompletion = [this](const Run* r) { runFinished(r); };
+                run->notifyCompletion = [this](Run* r) { runFinished(r); };
 
                 // trigger the first step of the run
                 if(stepRun(run)) {
                     // should never happen
-                    KJ_LOG(INFO, "No steps for run");
+                    LLOG(INFO, "No steps for run");
                     run->complete();
                 }
 
@@ -638,15 +666,25 @@ void Laminar::assignNewJobs() {
     }
 }
 
-void Laminar::runFinished(const Run * r) {
+void Laminar::runFinished(Run * r) {
     Node* node = r->node;
 
     node->busyExecutors--;
-    KJ_LOG(INFO, "Run completed", r->name, to_string(r->result));
+    LLOG(INFO, "Run completed", r->name, to_string(r->result));
     time_t completedAt = time(0);
-    db->stmt("INSERT INTO builds VALUES(?,?,?,?,?,?,?,?,?,?,?)")
+
+    // compress log
+    std::string zipped(r->log.size(), '\0');
+    size_t logsize = r->log.length();
+    size_t zippedSize = zipped.size();
+    ::compress((unsigned char*)&zipped[0], &zippedSize,
+            (unsigned char*)&r->log[0], logsize);
+    zipped.resize(zippedSize);
+
+    std::string reason = r->reason();
+    db->stmt("INSERT INTO builds VALUES(?,?,?,?,?,?,?,?,?,?,?,?)")
      .bind(r->name, r->build, node->name, r->queuedAt, r->startedAt, completedAt, int(r->result),
-           r->log, r->parentName, r->parentBuild, r->reason())
+           zipped, logsize, r->parentName, r->parentBuild, reason)
      .exec();
 
     // notify clients

src/laminar.h

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -36,7 +36,7 @@ class Json;
 // It owns a Server to manage the HTTP/websocket and Cap'n Proto RPC
 // interfaces and communicates via the LaminarInterface methods and
 // the LaminarClient objects (see interface.h)
-class Laminar : public LaminarInterface {
+class Laminar final : public LaminarInterface {
 public:
     Laminar();
     ~Laminar();
@@ -61,7 +61,7 @@ private:
     void reapAdvance();
     void assignNewJobs();
     bool stepRun(std::shared_ptr<Run> run);
-    void runFinished(const Run*);
+    void runFinished(Run*);
     bool nodeCanQueue(const Node&, const Run&) const;
     // expects that Json has started an array
     void populateArtifacts(Json& out, std::string job, int num) const;

src/log.h

@@ -0,0 +1,37 @@
+///
+/// Copyright 2015 Oliver Giles
+///
+/// This file is part of Laminar
+///
+/// Laminar is free software: you can redistribute it and/or modify
+/// it under the terms of the GNU General Public License as published by
+/// the Free Software Foundation, either version 3 of the License, or
+/// (at your option) any later version.
+///
+/// Laminar is distributed in the hope that it will be useful,
+/// but WITHOUT ANY WARRANTY; without even the implied warranty of
+/// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+/// GNU General Public License for more details.
+///
+/// You should have received a copy of the GNU General Public License
+/// along with Laminar.  If not, see <http://www.gnu.org/licenses/>
+///
+#ifndef _LAMINAR_LOG_H_
+#define _LAMINAR_LOG_H_
+
+#include <kj/debug.h>
+
+// Simple override to prevent full paths to source files from
+// appearing in log messages. Assumes / is the path separator.
+// @see kj/debug.h
+
+#define _LBASENAME(path) strrchr(path, '/') ? strrchr(path, '/') + 1 : path
+
+#define LLOG(severity, ...) \
+  if (!::kj::_::Debug::shouldLog(::kj::_::Debug::Severity::severity)) {} else \
+    ::kj::_::Debug::log(_LBASENAME(__FILE__), __LINE__, ::kj::_::Debug::Severity::severity, \
+                        #__VA_ARGS__, __VA_ARGS__)
+
+
+#endif // _LAMINAR_LOG_H_
+

src/main.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -17,12 +17,15 @@
 /// along with Laminar.  If not, see <http://www.gnu.org/licenses/>
 ///
 #include "laminar.h"
+#include "log.h"
 
 #include <signal.h>
-#include <kj/debug.h>
 
-std::function<void()> sigHandler;
+static Laminar* laminar;
-static void __sigHandler(int) { sigHandler(); }
+
+static void laminar_quit(int) {
+    laminar->stop();
+}
 
 int main(int argc, char** argv) {
     for(int i = 1; i < argc; ++i) {
@@ -31,19 +34,14 @@ int main(int argc, char** argv) {
         }
     }
 
-    do {
+    laminar = new Laminar;
-        Laminar laminar;
-        sigHandler = [&](){
-            KJ_LOG(INFO, "Received SIGINT");
-            laminar.stop();
-        };
-        signal(SIGINT, &__sigHandler);
-        signal(SIGTERM, &__sigHandler);
 
-        laminar.run();
+    signal(SIGINT, &laminar_quit);
-    } while(false);
+    signal(SIGTERM, &laminar_quit);
+    laminar->run();
 
-    KJ_DBG("end of main");
+    delete laminar;
+
+    LLOG(INFO, "Clean exit");
     return 0;
 }
-

src/resources.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///

src/resources.h

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///

src/resources/js/app.js

@@ -213,20 +213,20 @@ angular.module('laminar',['ngRoute','ngSanitize'])
 			$scope.$apply();
 			
 			var chtBt = new Chart(document.getElementById("chartBt").getContext("2d")).Bar({
-				labels: data.recent.map(function(e){return '#' + e.number;}),
+				labels: data.recent.map(function(e){return '#' + e.number;}).reverse(),
 				datasets: [{
 					fillColor: "darkseagreen",
 					strokeColor: "forestgreen",
-					data: data.recent.map(function(e){return e.duration;})
+					data: data.recent.map(function(e){return e.duration;}).reverse()
 				}]
 			},
 			{barValueSpacing: 1,barStrokeWidth: 1,barDatasetSpacing:0}
 			);
 
-			for(var i = 0; i < data.recent.length; ++i) {
+			for(var i = 0, n = data.recent.length; i < n; ++i) {
 				if(data.recent[i].result != "success") {
-					chtBt.datasets[0].bars[i].fillColor = "darksalmon";
+					chtBt.datasets[0].bars[n-i-1].fillColor = "darksalmon";
-					chtBt.datasets[0].bars[i].strokeColor = "crimson";
+					chtBt.datasets[0].bars[n-i-1].strokeColor = "crimson";
 				}
 			}
 			chtBt.update();
@@ -306,7 +306,10 @@ angular.module('laminar',['ngRoute','ngSanitize'])
 .run(function($rootScope) {
 	angular.extend($rootScope, {
 		runIcon: function(result) {
-			return result === "success" ? '<span style="color:forestgreen;font-family:\'Zapf Dingbats\';">✔</span>' : result === "failed" ? '<span style="color:crimson;">✘</span>' : '';
+			return result === "success" ? '<span style="color:forestgreen;font-family:\'Zapf Dingbats\';">✔</span>' : result === "failed" || result === "aborted" ? '<span style="color:crimson;">✘</span>' : '';
+		},
+		runComplete: function(run) {
+			return !!run && (run.result === 'aborted' || run.result === 'failed' || run.result === 'success');
 		},
 		formatDate: function(unix) {
 			// TODO reimplement when toLocaleDateString() accepts formatting

src/resources/tpl/run.html

@@ -1,10 +1,10 @@
 <div class="container-fluid">
 <div class="row">
 <div class="col-sm-5 col-md-6 col-lg-7">
- <h3 style="float:left"><img class="spin" src="/progress.png" ng-hide="job.result"><span ng-bind-html="runIcon(job.result)"></span> {{name}} #{{num}}</h3>
+ <h3 style="float:left"><img class="spin" src="/progress.png" ng-show="job.result === 'running'"><span ng-bind-html="runIcon(job.result)"></span> {{name}} #{{num}}</h3>
  <nav class="pull-left">
   <ul class="pagination" style="margin:15px 20px">
-   <li><a href="jobs/{{name}}/{{num-1}}">&laquo;</a></li>
+   <li ng-show="num > 1"><a href="jobs/{{name}}/{{num-1}}">&laquo;</a></li>
    <li ng-show="latestNum > num"><a ng-href="jobs/{{name}}/{{num+1}}">&raquo;</a></li>
   </ul>
  </nav>
@@ -13,12 +13,12 @@
   <dt>Reason</dt><dd>{{job.reason}}</dd>
   <dt>Queued for</dt><dd>{{job.queued}}s</dd>
   <dt>Started</dt><dd>{{formatDate(job.started)}}</dd>
-  <dt ng-show="job.result">Completed</dt><dd ng-show="job.result">{{formatDate(job.completed)}}</dd>
+  <dt ng-show="runComplete(job)">Completed</dt><dd ng-show="runComplete(job)">{{formatDate(job.completed)}}</dd>
-  <dt ng-show="job.result">Duration</dt><dd ng-show="job.result">{{job.duration}}s</dd>
+  <dt ng-show="runComplete(job)">Duration</dt><dd ng-show="runComplete(job)">{{job.duration}}s</dd>
  </dl>
 </div>
 <div class="col-sm-7 col-md-6 col-lg-5">
- <div class="progress" ng-hide="job.result">
+ <div class="progress" ng-show="job.result == 'running'">
   <div class="progress-bar progress-bar-{{job.overtime?'warning':'info'}} progress-bar-striped {{job.etc?'':'active'}}" style="width:{{!job.etc?'100':job.progress}}%;"></div>
  </div>
  <div class="panel panel-default" ng-show="job.artifacts.length">

src/run.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -19,10 +19,11 @@
 #include "run.h"
 #include "node.h"
 #include "conf.h"
-#include <iostream>
+#include "log.h"
-#include <kj/debug.h>
 
+#include <iostream>
 #include <unistd.h>
+#include <signal.h>
 
 #include <boost/filesystem.hpp>
 namespace fs = boost::filesystem;
@@ -30,6 +31,7 @@ namespace fs = boost::filesystem;
 std::string to_string(const RunState& rs) {
     switch(rs) {
     case RunState::PENDING: return "pending";
+    case RunState::RUNNING: return "running";
     case RunState::ABORTED: return "aborted";
     case RunState::FAILED: return "failed";
     case RunState::SUCCESS: return "success";
@@ -46,7 +48,7 @@ Run::Run() {
 }
 
 Run::~Run() {
-    KJ_DBG("Run destroyed");
+    LLOG(INFO, "Run destroyed");
 }
 
 std::string Run::reason() const {
@@ -67,7 +69,13 @@ bool Run::step() {
         int pfd[2];
         pipe(pfd);
         pid_t pid = fork();
-        if(pid == 0) {
+        if(pid == 0) { // child
+            // reset signal mask (SIGCHLD blocked in Laminar::start)
+            sigset_t mask;
+            sigemptyset(&mask);
+            sigaddset(&mask, SIGCHLD);
+            sigprocmask(SIG_UNBLOCK, &mask, NULL);
+
             close(pfd[0]);
             dup2(pfd[1], 1);
             dup2(pfd[1], 2);
@@ -100,13 +108,14 @@ bool Run::step() {
             for(auto& pair : params) {
                 setenv(pair.first.c_str(), pair.second.c_str(), false);
             }
-            printf("[laminar] Executing %s\n", currentScript.c_str());
+            fprintf(stderr, "[laminar] Executing %s\n", currentScript.c_str());
             execl(currentScript.c_str(), currentScript.c_str(), NULL);
-            KJ_LOG(FATAL, "execl returned", strerror(errno));
+            // cannot use LLOG because stdout/stderr are captured
+            fprintf(stderr, "[laminar] Failed to execute %s\n", currentScript.c_str());
             _exit(1);
         }
 
-        KJ_LOG(INFO, "Forked", currentScript, pid);
+        LLOG(INFO, "Forked", currentScript, pid);
         close(pfd[1]);
         fd = pfd[0];
         this->pid = pid;

src/run.h

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -29,6 +29,7 @@
 enum class RunState {
     UNKNOWN,
     PENDING,
+    RUNNING,
     ABORTED,
     FAILED,
     SUCCESS
@@ -67,7 +68,7 @@ public:
 
     std::string reason() const;
 
-    std::function<void(const Run*)> notifyCompletion;
+    std::function<void(Run*)> notifyCompletion;
     Node* node;
     RunState result;
     RunState lastResult;

src/server.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -20,12 +20,12 @@
 #include "interface.h"
 #include "laminar.capnp.h"
 #include "resources.h"
+#include "log.h"
 
 #include <capnp/ez-rpc.h>
 #include <capnp/rpc-twoparty.h>
 #include <capnp/rpc.capnp.h>
 #include <kj/async-io.h>
-#include <kj/debug.h>
 #include <kj/threadlocal.h>
 
 #include <websocketpp/config/core.hpp>
@@ -33,6 +33,10 @@
 
 #include <sys/eventfd.h>
 
+// Size of buffer used to read from file descriptors. Should be
+// a multiple of sizeof(struct signalfd_siginfo) == 128
+#define PROC_IO_BUFSIZE 4096
+
 // Configuration struct for the websocketpp template library.
 struct wsconfig : public websocketpp::config::core {
 //    static const websocketpp::log::level elog_level =
@@ -58,9 +62,9 @@ namespace {
 LaminarCi::JobResult fromRunState(RunState state) {
     switch(state) {
     case RunState::SUCCESS: return LaminarCi::JobResult::SUCCESS;
-    case RunState::FAILED: return LaminarCi::JobResult::FAILED;
+    case RunState::FAILED:  return LaminarCi::JobResult::FAILED;
+    case RunState::ABORTED: return LaminarCi::JobResult::ABORTED;
     default:
-        KJ_DBG("TODO log state", to_string(state));
         return LaminarCi::JobResult::UNKNOWN;
     }
 }
@@ -81,7 +85,7 @@ public:
     // Start a job, without waiting for it to finish
     kj::Promise<void> trigger(TriggerContext context) override {
         std::string jobName = context.getParams().getJobName();
-        KJ_LOG(INFO, "RPC trigger", jobName);
+        LLOG(INFO, "RPC trigger", jobName);
         ParamMap params;
         for(auto p : context.getParams().getParams()) {
             params[p.getName().cStr()] = p.getValue().cStr();
@@ -96,7 +100,7 @@ public:
     // Start a job and wait for the result
     kj::Promise<void> start(StartContext context) override {
         std::string jobName = context.getParams().getJobName();
-        KJ_LOG(INFO, "RPC start", jobName);
+        LLOG(INFO, "RPC start", jobName);
         ParamMap params;
         for(auto p : context.getParams().getParams()) {
             params[p.getName().cStr()] = p.getValue().cStr();
@@ -116,7 +120,7 @@ public:
     kj::Promise<void> pend(PendContext context) override {
         std::string jobName = context.getParams().getJobName();
         int buildNum = context.getParams().getBuildNum();
-        KJ_LOG(INFO, "RPC pend", jobName, buildNum);
+        LLOG(INFO, "RPC pend", jobName, buildNum);
 
         kj::Promise<RunState> promise = laminar.waitForRun(jobName, buildNum);
 
@@ -129,7 +133,7 @@ public:
     kj::Promise<void> set(SetContext context) override {
         std::string jobName = context.getParams().getJobName();
         int buildNum = context.getParams().getBuildNum();
-        KJ_LOG(INFO, "RPC set", jobName, buildNum);
+        LLOG(INFO, "RPC set", jobName, buildNum);
 
         LaminarCi::MethodResult result = laminar.setParam(jobName, buildNum,
             context.getParams().getParam().getName(), context.getParams().getParam().getValue())
@@ -371,22 +375,22 @@ void Server::start() {
     // this eventfd is just to allow us to quit the server at some point
     // in the future by adding this event to the async loop. I couldn't see
     // a simpler way...
-    efd = eventfd(0,0);
+    efd_quit = eventfd(0,0);
     kj::Promise<void> quit = kj::evalLater([this](){
         static uint64_t _;
-        auto wakeEvent = ioContext.lowLevelProvider->wrapInputFd(efd);
+        auto wakeEvent = ioContext.lowLevelProvider->wrapInputFd(efd_quit);
         return wakeEvent->read(&_, sizeof(uint64_t)).attach(std::move(wakeEvent));
     });
     quit.wait(ioContext.waitScope);
 }
 
 void Server::stop() {
-    eventfd_write(efd, 1);
+    eventfd_write(efd_quit, 1);
 }
 
-void Server::addProcess(int fd, std::function<void(char*,size_t)> readCb, std::function<void()> cb) {
+void Server::addDescriptor(int fd, std::function<void(char*,size_t)> cb) {
     auto event = this->ioContext.lowLevelProvider->wrapInputFd(fd);
-    tasks.add(handleProcessOutput(event,readCb).attach(std::move(event)).then(std::move(cb)));
+    tasks.add(handleFdRead(event, cb).attach(std::move(event)));
 }
 
 void Server::acceptHttpClient(kj::Own<kj::ConnectionReceiver>&& listener) {
@@ -416,14 +420,15 @@ void Server::acceptRpcClient(kj::Own<kj::ConnectionReceiver>&& listener) {
     );
 }
 
-// handles stdout/stderr from a child process by sending it to the provided
+// returns a promise which will read a chunk of data from the file descriptor
-// callback function
+// wrapped by stream and invoke the provided callback with the read data.
-kj::Promise<void> Server::handleProcessOutput(kj::AsyncInputStream* stream, std::function<void(char*,size_t)> readCb) {
+// Repeats until ::read returns <= 0
-    static char* buffer = new char[131072];
+kj::Promise<void> Server::handleFdRead(kj::AsyncInputStream* stream, std::function<void(char*,size_t)> cb) {
-    return stream->tryRead(buffer, 1, sizeof(buffer)).then([this,stream,readCb](size_t sz) {
+    static char* buffer = new char[PROC_IO_BUFSIZE];
-        readCb(buffer, sz);
+    return stream->tryRead(buffer, 1, PROC_IO_BUFSIZE).then([this,stream,cb](size_t sz) {
         if(sz > 0) {
-            return handleProcessOutput(stream, readCb);
+            cb(buffer, sz);
+            return handleFdRead(stream, cb);
         }
         return kj::Promise<void>(kj::READY_NOW);
     });

src/server.h

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -26,8 +26,8 @@
 
 struct LaminarInterface;
 
-// This class abstracts the HTTP/Websockets and Cap'n Proto RPC interfaces.
+// This class abstracts the HTTP/Websockets and Cap'n Proto RPC interfaces
-// It also manages the program's asynchronous event loop
+// and manages the program's asynchronous event loop
 class Server final : public kj::TaskSet::ErrorHandler {
 public:
     // Initializes the server with a LaminarInterface to handle requests from
@@ -38,22 +38,26 @@ public:
     ~Server();
     void start();
     void stop();
-    void addProcess(int fd, std::function<void(char*,size_t)> readCb, std::function<void()> cb);
+
+    // add a file descriptor to be monitored for output. The callback will be
+    // invoked with the read data
+    void addDescriptor(int fd, std::function<void(char*,size_t)> cb);
 
 private:
     void acceptHttpClient(kj::Own<kj::ConnectionReceiver>&& listener);
     void acceptRpcClient(kj::Own<kj::ConnectionReceiver>&& listener);
-    kj::Promise<void> handleProcessOutput(kj::AsyncInputStream* stream, std::function<void(char*,size_t)> readCb);
+    kj::Promise<void> handleFdRead(kj::AsyncInputStream* stream, std::function<void(char*,size_t)> cb);
 
     void taskFailed(kj::Exception&& exception) override {
         kj::throwFatalException(kj::mv(exception));
     }
 
 private:
-    int efd;
-    capnp::Capability::Client rpcInterface;
     struct WebsocketConnection;
     struct HttpImpl;
+
+    int efd_quit;
+    capnp::Capability::Client rpcInterface;
     HttpImpl* httpInterface;
     kj::AsyncIoContext ioContext;
     kj::TaskSet tasks;