Merge branch 'master' of github.com:mapbox/tilemill

commands/export.bones

@@ -108,12 +108,6 @@ command.prototype.initialize = function(plugin, callback) {
     // Set process title.
     process.title = 'tm-' + path.basename(opts.filepath);
 
-    // Catch SIGINT.
-    process.on('SIGINT', function () {
-        console.log('Got SIGINT. Run "kill ' + process.pid + '" to terminate.');
-    });
-    process.on('SIGUSR1', process.exit);
-
     // Upload format does not require loaded project.
     if (opts.format === 'upload') return this[opts.format](callback);
 

lib/queue.js

@@ -0,0 +1,46 @@
+var util = require('util');
+var EventEmitter = require('events').EventEmitter;
+
+module.exports = Queue;
+function Queue(callback, concurrency, timeout) {
+    this.callback = callback;
+    this.concurrency = concurrency || 10;
+    this.timeout = timeout || 0;
+    this.add = this.add.bind(this);
+    this.next = this.next.bind(this);
+    this.invoke = this.invoke.bind(this);
+    this.queue = [];
+    this.running = 0;
+}
+util.inherits(Queue, EventEmitter);
+
+Queue.prototype.add = function(item) {
+    this.queue.push(item);
+    if (this.running < this.concurrency) {
+        this.running++;
+        this.next();
+    }
+};
+
+Queue.prototype.invoke = function() {
+    if (this.queue.length) {
+        this.callback(this.queue.shift(), this.next);
+    } else {
+        this.next();
+    }
+};
+
+Queue.prototype.next = function(err) {
+    if (this.queue.length) {
+        if (this.timeout) {
+            setTimeout(this.invoke, this.timeout);
+        } else {
+            process.nextTick(this.invoke);
+        }
+    } else {
+        this.running--;
+        if (!this.running) {
+            this.emit('empty');
+        }
+    }
+};

models/Export.bones

@@ -63,12 +63,9 @@ model.prototype.schema = {
 };
 
 model.prototype.initialize = function() {
-    if (this.isNew()){
-        this.set({
-            created: +new Date,
-            id: (+new Date) + ''
-        }, {silent: true});
-    }
+    if (this.isNew()) this.set({
+        id: Date.now().toString()
+    }, {silent: true});
 };
 
 model.prototype.url = function() {

models/Exports.server.bones

@@ -1,16 +1,25 @@
 var Step = require('step'),
+    Queue = require('../lib/queue'),
     fs = require('fs'),
     path = require('path'),
     exec = require('child_process').exec,
     spawn = require('child_process').spawn,
-    settings = Bones.plugin.config;
+    settings = Bones.plugin.config,
+    pids = {};
+
+// Queue exports with concurrency 4.
+// @TODO make concurrency configurable?
+var queue = new Queue(start, 4);
+
+function start(id, callback) {
+    var model = new models.Export({id:id});
+    Step(function() {
+        Backbone.sync('read', model,
+            function(data) { this(null, data) }.bind(this),
+            function(err) { this(err) }.bind(this))
+    }, function(err, data) {
+        if (err || data.status !== 'waiting') return callback();
 
-// @TODO: need a queue system. Difficult to manage atm because process
-// completion is now determined outside this process.
-// See http://en.wikipedia.org/wiki/SIGCHLD ... may be useful for determining
-// when a child process has died and can be removed from the pool.
-var start = function(model, data, callback) {
-    if (data.status === 'waiting') {
         var args = [];
         // nice the export process.
         args.push('-n19');
@@ -37,72 +46,73 @@ var start = function(model, data, callback) {
         if (data.minzoom) args.push('--minzoom=' + data.minzoom);
         if (data.maxzoom) args.push('--maxzoom=' + data.maxzoom);
 
-        var options = {
-            env: process.env,
+        var child = spawn('nice', args, {
+            env: _(process.env).extend({
+                tilemillConfig:JSON.stringify(settings)
+            }),
             cwd: undefined,
             customFds: [-1, -1, -1],
             setsid: false
-        };
-        options.env.tilemillConfig = JSON.stringify(settings);
-        var child = spawn('nice', args, options);
-        model.set({pid:child.pid, status:'processing'});
-        Backbone.sync('update', model, callback, callback);
-    } else if (data.status === 'processing') {
-        var pid = data.pid || 0;
-        exec('ps -p ' + pid + ' | grep ' + pid, function(err, stdout) {
-            if (!err) return callback();
-            model.set({status: 'error', error: 'Export process died.'});
-            Backbone.sync('update', model, callback, callback);
         });
-    } else {
-        callback();
-    }
+        child.on('exit', callback);
+        pids[child.pid] = true;
+        (new models.Export(data)).save({
+            pid:child.pid,
+            created:Date.now(),
+            status:'processing'
+        });
+    });
+};
+
+// Export child processes are managed from the parent:
+// 1. when an export model with status 'waiting' is created or read
+//    it is queued for export.
+// 2. when reading models the process health is checked. if the pid
+//    is not found, the model's status should be updated.
+function check(data) {
+    if (data.status === 'processing' && data.pid && !pids[data.pid])
+        return { status: 'error', error: 'Export process died' };
+    if (data.status === 'waiting' && !_(queue.queue).include(data.id))
+        queue.add(data.id);
 };
 
 models.Export.prototype.sync = function(method, model, success, error) {
     switch (method) {
-    // Export child processes are managed from the parent:
-    // 1. when an export model is first created a process is started.
-    // 2. when reading models the process health is checked.
     case 'read':
     case 'create':
+    case 'update':
+        if (!model.id) throw new Error('Model ID is required.');
         Backbone.sync(method, model, function(data) {
-            start(model, model.toJSON(), function() {
-                success(data);
-            });
+            var attr = check(model.toJSON());
+            if (!attr) return success(data);
+
+            // If attributes are set we must further update this model
+            // to reflect its process status (it's dead, basically).
+            model.set(attr);
+            Backbone.sync('update', model, function() {
+                success(_(data).extend(attr));
+            }, error);
         }, error);
         break;
-    // Updates occur via the child process.
-    case 'update':
-        Backbone.sync(method, model, success, error);
-        break;
-    // Deletion kills the child process and removes the export file if it
-    // exists. Note that SIGUSR1 is used instead of SIGINT for two reasons:
-    // 1. The child process does not exit directly on SIGINT to prevent it
-    //    from going down if the parent goes down.
-    // 2. If the model `pid` is stale and somehow the `pid` is now occupied
-    //    by another process SIGUSR1 likely won't kill the process.
+    // Deletion kills the child process and removes the export file.
     case 'delete':
         Step(function() {
             Backbone.sync('read', model, this, this);
-        },
-        function(data) {
-            if (data && data.pid) {
-                // Try/catch as process may not exist.
-                try { process.kill(data.pid, 'SIGUSR1'); }
+        }, function(data) {
+            // Try/catch as process may not exist.
+            if (data && data.pid && pids[data.pid]) {
+                delete pids[data.pid];
+                // try { process.kill(data.pid, 'SIGUSR1'); }
+                try { process.kill(data.pid, 'SIGINT'); }
                 catch(err) {}
             }
             if (data && data.filename && data.format !== 'upload') {
-                var filepath = path.join(settings.files, 'export', data.filename);
-                path.exists(filepath, function(exists) {
-                    if (exists) return fs.unlink(filepath, this);
-                    this();
-                }.bind(this));
+                fs.unlink(path.join(settings.files, 'export', data.filename), this);
             } else {
                 this();
             }
-        },
-        function(err) {
+        }, function(err) {
+            if (err && err.code !== 'ENOENT') return error(err);
             Backbone.sync(method, model, success, error);
         });
         break;
@@ -111,15 +121,16 @@ models.Export.prototype.sync = function(method, model, success, error) {
 
 models.Exports.prototype.sync = function(method, model, success, error) {
     if (method !== 'read') return success({});
-    Backbone.sync(method, model, function(data) {
-        Step(function() {
-            var group = this.group();
-            _(data).each(function(m) {
-                var model = new models.Export(m);
-                start(model, model.toJSON(), group());
-            });
-        }, function() {
-            success(data);
-        });
-    }, error);
+    Step(function() {
+        Backbone.sync(method, model, this, error);
+    }, function(data) {
+        if (!data || !data.length) return this(null, []);
+        Bones.utils.fetch(data.reduce(function(memo, d) {
+            memo[d.id] = new models.Export(d);
+            return memo;
+        }, {}), this);
+    }, function(err, models) {
+        if (err) return error(err);
+        success(_(models).map(function(m) { return m.toJSON() }));
+    });
 };

models/Preview.server.bones

@@ -9,7 +9,7 @@ models.Preview.prototype.sync = function(method, model, success, error) {
         if (err) return error(err);
         source.getInfo(function(err, info) {
             if (err) return error(err);
-            info.tiles = ['/tile/' + model.id + '/{z}/{x}/{y}.png'];
+            info.tiles = ['http://' + settings.tileUrl + '/tile/' + model.id + '/{z}/{x}/{y}.png'];
             success(_(info).extend({id: model.id }));
         });
     });

templates/Exports._

@@ -11,7 +11,11 @@
   </span>
   <h2><%= m.get('filename') %></h2>
   <div class='description'>
-    Started <%= (new Date(m.get('created'))).format('F j h:ia') %>
+    <% if (m.get('created')) { %>
+    Started <%= (new Date(m.get('created'))).format('M j h:ia') %>
+    <% } else if (m.get('status') === 'waiting') { %>
+    Waiting to be processed
+    <% } %>
     <% if (m.get('updated')) { %>
     &mdash; <%= obj.time(m.get('updated') - m.get('created')) %>
     <% } %>

test/abilities.test.js

@@ -1,6 +1,8 @@
 var assert = require('assert');
 
 require('./support/start')(function(command) {
+    command.servers['Tile'].close();
+
     exports['test abilities endpoint'] = function() {
         assert.response(command.servers['Core'],
             { url: '/assets/tilemill/js/abilities.js' },

test/config.test.js

@@ -2,6 +2,8 @@ var assert = require('assert');
 var Step = require('step');
 
 require('./support/start')(function(command) {
+command.servers['Tile'].close();
+
 exports['config'] = function() {
 
 var server = command.servers['Core'];

test/datasource.test.js

@@ -8,6 +8,7 @@ function readJSON(name) {
 }
 
 require('./support/start')(function(command) {
+    command.servers['Core'].close();
 
     exports['test sqlite datasource'] = function() {
         assert.response(command.servers['Tile'],

test/export.test.js

@@ -8,14 +8,13 @@ function readJSON(name) {
 }
 
 require('./support/start')(function(command) {
+    command.servers['Tile'].close();
 
     exports['test export job creation'] = function(beforeExit) {
         var completed = false;
-        var created = Date.now()
-        var id = String(created);
+        var id = Date.now().toString();
         var job = readJSON('export-job');
         var token = job['bones.token'];
-        job.created = created;
         job.id = id;
 
         assert.response(command.servers['Core'], {
@@ -35,10 +34,12 @@ require('./support/start')(function(command) {
             }, { status: 200 }, function(res) {
                 var body = JSON.parse(res.body);
                 job.status = "processing";
-                delete job['bones.token'];
                 assert.ok(body[0].pid);
+                assert.ok(body[0].created);
+                delete job['bones.token'];
+                delete body[0].created;
                 delete body[0].pid;
-                assert.deepEqual([job], body);
+                assert.deepEqual(job, body[0]);
 
                 job['bones.token'] = token;
                 assert.response(command.servers['Core'], {

test/fixtures/export-job.json

@@ -4,11 +4,10 @@
     "format": "mbtiles",
     "project": "demo_01",
     "tile_format": "png",
-    "created": 0,
     "id": "",
     "filename": "demo_01.mbtiles",
     "bbox": [ -29.5642, 31.8402, 42.1545, 71.9245 ],
     "minzoom": 0,
-    "maxzoom": 9,
+    "maxzoom": 2,
     "bones.token": "zbx6dr0ghgvRNZOuu8PXtt2VCAIKO2qK"
 }

test/project.test.js

@@ -19,6 +19,8 @@ function cleanProject(proj) {
 }
 
 require('./support/start')(function(command) {
+    command.servers['Tile'].close();
+
     exports['test project collection endpoint'] = function() {
         assert.response(command.servers['Core'],
             { url: '/api/Project' },