v0.3

CMakeLists.txt

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

docker-build-centos.sh

@@ -1,6 +1,6 @@
 #!/bin/bash
 
-VERSION=0.2
+VERSION=0.3
 
 OUTPUT_DIR=$PWD
 

docker-build-debian.sh

@@ -1,12 +1,12 @@
 #!/bin/bash
 
-VERSION=0.2
+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 ubuntu bash -xe <<EOS
+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
@@ -67,5 +67,5 @@ EOF
 chmod +x laminar/DEBIAN/postinst
 
 dpkg-deb --build laminar
-mv laminar.deb /output/laminar_$VERSION-1_amd64.deb
+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
 ///
@@ -50,14 +50,14 @@ 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) {
-    sqlite3_bind_blob(stmt, i, e.data(), e.size(), NULL);
+void Database::Statement::bindValue(int i, const std::string& e) {
+    sqlite3_bind_text(stmt, i, e.data(), e.size(), NULL);
 }
 
 template<> std::string Database::Statement::fetchColumn(int col) {
     int sz = sqlite3_column_bytes(stmt, col);
     std::string res(sz, '\0');
-    memcpy(&res[0], sqlite3_column_blob(stmt, col), sz);
+    memcpy(&res[0], sqlite3_column_text(stmt, col), sz);
     return res;
 }
 

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
 ///
@@ -22,6 +22,7 @@
 #include "log.h"
 
 #include <sys/wait.h>
+#include <sys/signalfd.h>
 #include <fstream>
 #include <zlib.h>
 
@@ -326,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();
 }
 
@@ -474,25 +488,24 @@ 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
-    // we re-schedule a poll to wait for finished child processes?
-    pid_t pid = waitpid(-1, &ret, 0);
-    // TODO: handle signalled child processes
-    if(pid > 0) {
+    pid_t pid;
+    while((pid = waitpid(-1, &ret, WNOHANG)) > 0) {
         LLOG(INFO, "Reaping", pid);
         auto it = activeJobs.get<0>().find(pid);
         std::shared_ptr<Run> run = *it;
@@ -505,7 +518,6 @@ void Laminar::reapAdvance() {
         if(completed)
             run->complete();
     }
-    assignNewJobs();
 }
 
 bool Laminar::nodeCanQueue(const Node& node, const Run& run) const {
@@ -669,9 +681,10 @@ void Laminar::runFinished(Run * r) {
             (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),
-           zipped, logsize, 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();

src/main.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -21,8 +21,11 @@
 
 #include <signal.h>
 
-std::function<void()> sigHandler;
-static void __sigHandler(int) { sigHandler(); }
+static Laminar* laminar;
+
+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 laminar;
-        sigHandler = [&](){
-            LLOG(INFO, "Received SIGINT");
-            laminar.stop();
-        };
-        signal(SIGINT, &__sigHandler);
-        signal(SIGTERM, &__sigHandler);
+    laminar = new Laminar;
 
-        laminar.run();
-    } while(false);
+    signal(SIGINT, &laminar_quit);
+    signal(SIGTERM, &laminar_quit);
+    laminar->run();
 
-    LLOG(INFO, "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/run.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -23,6 +23,7 @@
 
 #include <iostream>
 #include <unistd.h>
+#include <signal.h>
 
 #include <boost/filesystem.hpp>
 namespace fs = boost::filesystem;
@@ -68,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);

src/run.h

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

src/server.cpp

@@ -1,5 +1,5 @@
 ///
-/// Copyright 2015 Oliver Giles
+/// Copyright 2016 Oliver Giles
 ///
 /// This file is part of Laminar
 ///
@@ -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 =
@@ -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
-// callback function
-kj::Promise<void> Server::handleProcessOutput(kj::AsyncInputStream* stream, std::function<void(char*,size_t)> readCb) {
-    static char* buffer = new char[131072];
-    return stream->tryRead(buffer, 1, sizeof(buffer)).then([this,stream,readCb](size_t sz) {
-        readCb(buffer, sz);
+// returns a promise which will read a chunk of data from the file descriptor
+// wrapped by stream and invoke the provided callback with the read data.
+// Repeats until ::read returns <= 0
+kj::Promise<void> Server::handleFdRead(kj::AsyncInputStream* stream, std::function<void(char*,size_t)> cb) {
+    static char* buffer = new char[PROC_IO_BUFSIZE];
+    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.
-// It also manages the program's asynchronous event loop
+// This class abstracts the HTTP/Websockets and Cap'n Proto RPC interfaces
+// 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;