v0.16.0

* WIP #824 Added extra logs

* WIP #824 Moved creation events to the local diff strategy

* WIP #824 Added deletion event to the local stored changes to apply

* WIP #824 Fixed variable name

* WIP #824 Fixed apply dependent creation changes

* WIP #824 Update local operation setAttr, getAttr

* WIP #824 Updated execpipeline save method

* WIP #824 Fixed code climate issue

I will probably remove all ArtifactFinder fn-ality

* WIP #824 Removed unused var

* WIP #824 update the meta on fast forward

.npmignore

@@ -1,40 +0,0 @@
-src/worker/tmp
-.env
-*.swp
-*.swo
-**.sass-cache
-# Logs
-logs
-*.log
-
-# Runtime data
-pids
-*.pid
-*.seed
-
-# Directory for instrumented libs generated by jscoverage/JSCover
-lib-cov
-
-# Coverage directory used by tools like istanbul
-coverage
-
-# Grunt intermediate storage (http://gruntjs.com/creating-plugins#storing-task-files)
-.grunt
-
-# node-waf configuration
-.lock-wscript
-
-# Compiled binary addons (http://nodejs.org/api/addons.html)
-build/Release
-
-# Dependency directory
-# https://www.npmjs.org/doc/misc/npm-faq.html#should-i-check-my-node_modules-folder-into-git
-node_modules
-tmp/
-test-tmp/
-blob-local-storage/
-src/seeds/nn/hash.txt
-src/seeds/pipeline/hash.txt
-
-# npm specific things
-images/

package.json

@@ -12,7 +12,7 @@
     "watch-test": "./node_modules/nodemon/bin/nodemon.js --exec 'node ./node_modules/mocha/bin/mocha --recursive test'",
     "build-nn": "node ./utils/nn-parser.js"
   },
-  "version": "0.15.0",
+  "version": "0.16.0",
   "dependencies": {
     "commander": "^2.9.0",
     "dotenv": "^2.0.0",

src/common/plugin/LocalExecutor.js

@@ -20,13 +20,13 @@ define([
                 var output = cntrs
                     .find(cntr => {
                         var metaNode = this.core.getMetaType(cntr),
-                            metaName = this.core.getAttribute(metaNode, 'name');
+                            metaName = this.getAttribute(metaNode, 'name');
                         return metaName === 'Outputs';
                     });
                 return this.core.loadChildren(output);
             })
             .then(dataNodes => {
-                hash = this.core.getAttribute(dataNodes[0], 'data');
+                hash = this.getAttribute(dataNodes[0], 'data');
                 return this.getOutputs(node);
             })
             .then(outputTuples => {
@@ -48,7 +48,7 @@ define([
         var hash,
             typeId = this.core.getPointerPath(node, 'type'),
             type,
-            artifactName = this.core.getAttribute(node, 'artifactName');
+            artifactName = this.getAttribute(node, 'artifactName');
 
         return this.core.loadByPath(this.rootNode, typeId)
             .then(_type => {
@@ -58,25 +58,20 @@ define([
             .then(saveDir => this.core.loadChildren(saveDir))
             .then(artifacts => {
                 return artifacts.find(artifact =>
-                    this.core.getAttribute(artifact, 'name') === artifactName &&
+                    this.getAttribute(artifact, 'name') === artifactName &&
                         this.isMetaTypeOf(artifact, type));
             })
             .then(matchingArtifact => {
-                hash = matchingArtifact && this.core.getAttribute(matchingArtifact, 'data');
+                hash = matchingArtifact && this.getAttribute(matchingArtifact, 'data');
                 // If no hash, just continue (the subsequent ops will receive 'nil')
                 if (!hash) {
                     return this.onOperationComplete(node);
                 } else {
                     return this.getOutputs(node)
                         .then(outputPairs => {
-                            var outputs = outputPairs.map(pair => pair[2]),
-                                paths;
-
-                            paths = outputs.map(output => this.core.getPath(output));
+                            var outputs = outputPairs.map(pair => pair[2]);
                             // Get the 'data' hash and store it in the output data ports
-                            this.logger.info(`Loading blob data (${hash}) to ${paths.map(p => `"${p}"`)}`);
-
-                            outputs.forEach(output => this.core.setAttribute(output, 'data', hash));
+                            outputs.forEach(output => this.setAttribute(output, 'data', hash));
 
                             this.onOperationComplete(node);
                         });
@@ -99,7 +94,7 @@ define([
 
                 if (containers.length > 1) {
                     saveDir = containers.find(c =>
-                        this.core.getAttribute(c, 'name').toLowerCase().indexOf('artifacts') > -1
+                        this.getAttribute(c, 'name').toLowerCase().indexOf('artifacts') > -1
                     ) || containers[0];
                 }
 
@@ -121,8 +116,8 @@ define([
             .then(artifacts => {
                 currNameHashPairs = artifacts
                     .map(node => [
-                        this.core.getAttribute(node, 'name'),
-                        this.core.getAttribute(node, 'data')
+                        this.getAttribute(node, 'name'),
+                        this.getAttribute(node, 'data')
                     ]);
                 return this.getInputs(node);
             })
@@ -142,9 +137,9 @@ define([
 
                 // Remove nodes that already exist
                 dataNodes = allDataNodes.filter(dataNode => {
-                    var hash = this.core.getAttribute(dataNode, 'data'),
+                    var hash = this.getAttribute(dataNode, 'data'),
                         name = this.core.getOwnAttribute(node, 'saveName') ||
-                            this.core.getAttribute(dataNode, 'name');
+                            this.getAttribute(dataNode, 'name');
 
                     return !(currNameHashPairs
                         .find(pair => pair[0] === name && pair[1] === hash));
@@ -156,10 +151,10 @@ define([
                         newName = this.core.getOwnAttribute(node, 'saveName');
                     if (newName) {
                         newNodes.forEach(node =>
-                            this.core.setAttribute(node, 'name', newName)
+                            this.setAttribute(node, 'name', newName)
                         );
                     }
-                    var hashes = dataNodes.map(n => this.core.getAttribute(n, 'data'));
+                    var hashes = dataNodes.map(n => this.getAttribute(n, 'data'));
                     this.logger.info(`saving hashes: ${hashes.map(h => `"${h}"`)}`);
                 } else if (allDataNodes.length === 0) {
                     this.logger.warn('No data nodes found!');

src/plugins/ExecuteJob/ExecuteJob.js

@@ -2,6 +2,7 @@
 /*jshint node:true, browser:true*/
 
 define([
+    'common/util/assert',
     'common/storage/constants',
     'text!./metadata.json',
     'executor/ExecutorClient',
@@ -15,6 +16,7 @@ define([
     'q',
     'underscore'
 ], function (
+    assert,
     STORAGE_CONSTANTS,
     pluginMetadata,
     ExecutorClient,
@@ -33,7 +35,9 @@ define([
     pluginMetadata = JSON.parse(pluginMetadata);
 
     var OUTPUT_INTERVAL = 1500,
-        STDOUT_FILE = 'job_stdout.txt';
+        STDOUT_FILE = 'job_stdout.txt',
+        CREATE_PREFIX = 'created_node_',
+        INDEX = 1;
 
     /**
      * Initializes a new instance of ExecuteJob.
@@ -54,6 +58,9 @@ define([
         this.lastAppliedCmd = {};
         this.canceled = false;
         this.changes = {};
+        this.creations = {};
+        this.deletions = [];
+        this.createIdToMetadataId = {};
         this.logManager = null;
     };
 
@@ -117,12 +124,57 @@ define([
         });
     };
 
+    //////////////////////////// Safe Save ////////////////////////////
+    ExecuteJob.prototype.getCreateId = function () {
+        return CREATE_PREFIX + (++INDEX);
+    };
+
+    ExecuteJob.prototype.isCreateId = function (id) {
+        return (typeof id === 'string') && (id.indexOf(CREATE_PREFIX) === 0);
+    };
+
+    ExecuteJob.prototype.createNode = function (baseType, parent) {
+        var id = this.getCreateId(),
+            parentId;
+
+        if (this.isCreateId(parent)) {
+            parentId = parent;
+        } else {
+            parentId = this.core.getPath(parent);
+        }
+
+        this.logger.info(`Creating ${id} of type ${baseType} in ${parentId}`);
+        assert(this.META[baseType], `Cannot create node w/ unrecognized type: ${baseType}`);
+        this.creations[id] = {
+            base: baseType,
+            parent: parentId
+        };
+        return id;
+    };
+
+    ExecuteJob.prototype.deleteNode = function (nodeId) {
+        this.deletions.push(nodeId);
+    };
+
     ExecuteJob.prototype.delAttribute = function (node, attr) {
         return this.setAttribute(node, attr, null);
     };
 
     ExecuteJob.prototype.setAttribute = function (node, attr, value) {
-        var nodeId = this.core.getPath(node);
+        var nodeId;
+
+        if (this.isCreateId(node)) {
+            nodeId = node;
+        } else {
+            nodeId = this.core.getPath(node);
+            assert(typeof nodeId === 'string', `Cannot set attribute of ${nodeId}`);
+        }
+
+        if (value !== null) {
+            this.logger.info(`Setting ${attr} of ${nodeId} to ${value}`);
+        } else {
+            this.logger.info(`Deleting ${attr} of ${nodeId}`);
+        }
 
         if (!this.changes[nodeId]) {
             this.changes[nodeId] = {};
@@ -131,17 +183,30 @@ define([
     };
 
     ExecuteJob.prototype.getAttribute = function (node, attr) {
-        var nodeId = this.core.getPath(node),
-            base,
-            baseId;
+        var nodeId,
+            base;
+
+        assert(this.deletions.indexOf(nodeId) === -1,
+            `Cannot get ${attr} from deleted node ${nodeId}`);
+
+        // Check if it was newly created
+        if (this.isCreateId(node)) {
+            nodeId = node;
+            assert(this.creations[nodeId], `Creation node not updated: ${nodeId}`);
+
+            // Set the node to the base so it falls back to an
+            // existing node if the attr info isn't in the diff
+            node = this.META[this.creations[nodeId].base];
+        } else {
+            nodeId = this.core.getPath(node);
+        }
 
         // Check the changes; fallback on actual node
         if (this.changes[nodeId] && this.changes[nodeId][attr] !== undefined) {
             // If deleted the attribute, get the default (inherited) value
             if (this.changes[nodeId][attr] === null) {
                 base = this.core.getBase(node);
-                baseId = this.core.getPath(base);
-                return this.getAttribute(baseId, attr);
+                return this.getAttribute(base, attr);
             }
             return this.changes[nodeId][attr];
         }
@@ -153,6 +218,7 @@ define([
         var attr,
             value;
 
+        this.logger.info(`About to apply changes for ${this.core.getPath(node)}`);
         for (var i = changes.length; i--;) {
             attr = changes[i][0];
             value = changes[i][1];
@@ -165,6 +231,12 @@ define([
         return node;
     };
 
+    ExecuteJob.prototype.applyModelChanges = function () {
+        return this.applyCreations()
+            .then(() => this.applyChanges())
+            .then(() => this.applyDeletions());
+    };
+
     ExecuteJob.prototype.applyChanges = function () {
         var nodeIds = Object.keys(this.changes),
             attrs,
@@ -175,6 +247,7 @@ define([
             id,
             promise;
 
+        this.logger.info('Collecting changes to apply in commit');
         for (var i = nodeIds.length; i--;) {
             changes = [];
             attrs = Object.keys(this.changes[nodeIds[i]]);
@@ -183,27 +256,160 @@ define([
                 changes.push([attrs[a], value]);
             }
             changesFor[nodeIds[i]] = changes;
+
+            assert(changes, `changes are invalid for ${nodeIds[i]}: ${changes}`);
+            assert(!this.isCreateId(nodeIds[i]),
+                `Creation id not resolved to actual id: ${nodeIds[i]}`);
             promise = this.core.loadByPath(this.rootNode, nodeIds[i]);
             promises.push(promise);
         }
 
         this.changes = {};
+        this.logger.info(`About to apply changes for ${promises.length} nodes`);
+        return Q.all(promises)
+            .then(nodes => {
+                for (var i = nodes.length; i--;) {
+                    id = this.core.getPath(nodes[i]);
+                    assert(nodes[i], `node is ${nodes[i]} (${nodeIds[i]})`);
+                    this._applyNodeChanges(nodes[i], changesFor[id]);
+                }
+            });
+    };
+
+    ExecuteJob.prototype.applyCreations = function () {
+        var nodeIds = Object.keys(this.creations),
+            tiers = this.createCreationTiers(nodeIds),
+            creations = this.creations,
+            newIds = {},
+            promise = Q(),
+            tier;
+
+        this.logger.info('Applying node creations');
+        for (var i = 0; i < tiers.length; i++) {
+            tier = tiers[i];
+            // Chain the promises, loading each tier sequentially
+            promise = promise.then(this.applyCreationTier.bind(this, creations, newIds, tier));
+        }
+
+        this.creations = {};
+        return promise;
+    };
+
+    ExecuteJob.prototype.applyCreationTier = function (creations, newIds, tier) {
+        var promises = [],
+            parentId,
+            node;
+
+        for (var j = tier.length; j--;) {
+            node = creations[tier[j]];
+            assert(node, `Could not find create info for ${tier[j]}`);
+            parentId = newIds[node.parent] || node.parent;
+            promises.push(this.applyCreation(tier[j], node.base, parentId));
+        }
         return Q.all(promises).then(nodes => {
-            for (var i = nodes.length; i--;) {
-                id = this.core.getPath(nodes[i]);
-                this._applyNodeChanges(nodes[i], changesFor[id]);
+            // Record the newIds so they can be used to resolve creation ids
+            // in subsequent tiers
+            for (var i = tier.length; i--;) {
+                newIds[tier[i]] = this.core.getPath(nodes[i]);
             }
         });
     };
 
+    // Figure out the dependencies between nodes to create.
+    // eg, if newId1 is to be created in newId2, then newId2 will
+    // be in an earlier tier than newId1. Essentially a topo-sort
+    // on a tree structure
+    ExecuteJob.prototype.createCreationTiers = function (nodeIds) {
+        var tiers = [],
+            prevTier = {},
+            tier = {},
+            id,
+            prevLen,
+            i;
+
+        // Create first tier (created inside existing nodes)
+        for (i = nodeIds.length; i--;) {
+            id = nodeIds[i];
+            if (!this.isCreateId(this.creations[id].parent)) {
+                tier[id] = true;
+                nodeIds.splice(i, 1);
+            }
+        }
+        prevTier = tier;
+        tiers.push(Object.keys(tier));
+
+        // Now, each tier consists of the nodes to be created inside a
+        // node from the previous tier
+        while (nodeIds.length) {
+            prevLen = nodeIds.length;
+            tier = {};
+            for (i = nodeIds.length; i--;) {
+                id = nodeIds[i];
+                if (prevTier[this.creations[id].parent]) {
+                    tier[id] = true;
+                    nodeIds.splice(i, 1);
+                }
+            }
+            prevTier = tier;
+            tiers.push(Object.keys(tier));
+            // Every iteration should find at least one node
+            assert(prevLen > nodeIds.length,
+                `Created empty create tier! Remaining: ${nodeIds.join(', ')}`);
+        }
+
+        return tiers;
+    };
+
+    ExecuteJob.prototype.applyCreation = function (tmpId, baseType, parentId) {
+        var base = this.META[baseType],
+            nodeId,
+            id;
+
+        this.logger.info(`Applying creation of ${tmpId} (${baseType}) in ${parentId}`);
+
+        assert(!this.isCreateId(parentId),
+            `Did not resolve parent id: ${parentId} for ${tmpId}`);
+        assert(base, `Invalid base type: ${baseType}`);
+        return this.core.loadByPath(this.rootNode, parentId)
+            .then(parent => this.core.createNode({base, parent}))
+            .then(node => {  // Update the _metadata records
+                id = this.createIdToMetadataId[tmpId];
+                delete this.createIdToMetadataId[tmpId];
+                this._metadata[id] = node;
+
+                // Update creations
+                nodeId = this.core.getPath(node);
+                if (this.changes[tmpId]) {
+                    assert(!this.changes[nodeId],
+                        `Newly created node cannot already have changes! (${nodeId})`);
+                    this.changes[nodeId] = this.changes[tmpId];
+                    delete this.changes[tmpId];
+                }
+                return node;
+            });
+    };
+
+    ExecuteJob.prototype.applyDeletions = function () {
+        var deletions = this.deletions;
+
+        this.deletions = [];
+        return Q.all(deletions.map(id => this.core.loadByPath(this.rootNode, id)))
+            .then(nodes => {
+                for (var i = nodes.length; i--;) {
+                    this.core.deleteNode(nodes[i]);
+                }
+            });
+    };
+
     // Override 'save' to notify the user on fork
     ExecuteJob.prototype.save = function (msg) {
         var name = this.getAttribute(this.activeNode, 'name');
+
         return this.updateForkName(name)
-            .then(() => this.applyChanges())
+            .then(() => this.applyModelChanges())
             .then(() => PluginBase.prototype.save.call(this, msg))
             .then(result => {
-                var msg;
+                this.logger.info(`Save finished w/ status: ${result.status}`);
                 if (result.status === STORAGE_CONSTANTS.FORKED) {
                     msg = `"${name}" execution has forked to "${result.forkName}"`;
                     this.currentForkName = result.forkName;
@@ -228,9 +434,20 @@ define([
                 this.rootNode = rootObject;
                 return this.core.loadByPath(rootObject,activeId);
             })
-            .then(activeObject => this.activeNode = activeObject);
+            .then(activeObject => this.activeNode = activeObject)
+            .then(() => {
+                var metaNames = Object.keys(this.META);
+                return Q.all(metaNames.map(name => this.updateMetaNode(name)));
+            });
     };
 
+    ExecuteJob.prototype.updateMetaNode = function (name) {
+        var id = this.core.getPath(this.META[name]);
+        return this.core.loadByPath(this.rootNode, id).then(node => this.META[name] = node);
+    };
+
+    //////////////////////////// END Safe Save ////////////////////////////
+
     ExecuteJob.prototype.getConnections = function (nodes) {
         var conns = [];
         for (var i = nodes.length; i--;) {
@@ -279,7 +496,8 @@ define([
             idsToDelete = [],
             type,
             base,
-            child;
+            child,
+            i;
 
         this.lastAppliedCmd[nodeId] = 0;
         this._oldMetadataByName[nodeId] = {};
@@ -287,7 +505,7 @@ define([
         return this.core.loadChildren(job)
             .then(jobChildren => {
                 // Remove any metadata nodes
-                for (var i = jobChildren.length; i--;) {
+                for (i = jobChildren.length; i--;) {
                     child = jobChildren[i];
                     if (this.isMetaTypeOf(child, this.META.Metadata)) {
                         id = this.core.getPath(child);
@@ -310,18 +528,22 @@ define([
                 }
 
                 // make the deletion ids relative to the job node
-                idsToDelete = idsToDelete.map(id => id.replace(nodeId, ''));
-                return Q.all(idsToDelete.map(id => this.core.loadByPath(job, id)));
-            })
-            .then(nodes => nodes.forEach(node => this.core.deleteNode(node)));
+                this.logger.debug(`About to delete ${idsToDelete.length}: ${idsToDelete.join(', ')}`);
+                for (i = idsToDelete.length; i--;) {
+                    this.deleteNode(idsToDelete[i]);
+                }
+            });
     };
 
     ExecuteJob.prototype.clearOldMetadata = function (job) {
         var nodeId = this.core.getPath(job),
-            nodeIds = Object.keys(this._markForDeletion[nodeId]);
+            nodeIds = Object.keys(this._markForDeletion[nodeId]),
+            node;
 
+        this.logger.debug(`About to delete ${nodeIds.length}: ${nodeIds.join(', ')}`);
         for (var i = nodeIds.length; i--;) {
-            this.core.deleteNode(this._markForDeletion[nodeId][nodeIds[i]]);
+            node = this._markForDeletion[nodeId][nodeIds[i]];
+            this.deleteNode(this.core.getPath(node));
         }
         delete this.lastAppliedCmd[nodeId];
         delete this._markForDeletion[nodeId];
@@ -1186,14 +1408,12 @@ define([
         // Check if the graph already exists
         graph = this._getExistingMetadata(jobId, 'Graph', name);
         if (!graph) {
-            graph = this.core.createNode({
-                base: this.META.Graph,
-                parent: job
-            });
+            graph = this.createNode('Graph', job);
 
             if (name) {
                 this.setAttribute(graph, 'name', name);
             }
+            this.createIdToMetadataId[graph] = id;
         }
 
         this._metadata[id] = graph;
@@ -1201,25 +1421,25 @@ define([
 
     ExecuteJob.prototype[CONSTANTS.GRAPH_PLOT] = function (job, id, x, y) {
         var jobId = this.core.getPath(job),
-            nonNum = /[^\d\.]*/g,
-            graph,
+            nonNum = /[^\d-\.]*/g,
+            line,
             points;
             
 
         id = jobId + '/' + id;
         this.logger.info(`Adding point ${x}, ${y} to ${id}`);
-        graph = this._metadata[id];
-        if (!graph) {
-            this.logger.warn(`Can't add point to non-existent graph: ${id}`);
+        line = this._metadata[id];
+        if (!line) {
+            this.logger.warn(`Can't add point to non-existent line: ${id}`);
             return;
         }
 
         // Clean the points by removing and special characters
         x = x.replace(nonNum, '');
         y = y.replace(nonNum, '');
-        points = this.getAttribute(graph, 'points');
+        points = this.getAttribute(line, 'points');
         points += `${x},${y};`;
-        this.setAttribute(graph, 'points', points);
+        this.setAttribute(line, 'points', points);
     };
 
     ExecuteJob.prototype[CONSTANTS.GRAPH_CREATE_LINE] = function (job, graphId, id) {
@@ -1230,12 +1450,10 @@ define([
 
         // Create a 'line' node in the given Graph metadata node
         name = name.replace(/\s+$/, '');
-        line = this.core.createNode({
-            base: this.META.Line,
-            parent: graph
-        });
+        line = this.createNode('Line', graph);
         this.setAttribute(line, 'name', name);
         this._metadata[jobId + '/' + id] = line;
+        this.createIdToMetadataId[line] = jobId + '/' + id;
     };
 
     ExecuteJob.prototype[CONSTANTS.IMAGE.BASIC] =
@@ -1265,11 +1483,9 @@ define([
             imageNode = this._getExistingMetadata(jobId, 'Image', name);
             if (!imageNode) {
                 this.logger.info(`Creating image ${id} named ${name}`);
-                imageNode = this.core.createNode({
-                    base: this.META.Image,
-                    parent: job
-                });
+                imageNode = this.createNode('Image', job);
                 this.setAttribute(imageNode, 'name', name);
+                this.createIdToMetadataId[imageNode] = id;
             }
             this._metadata[id] = imageNode;
         }

src/plugins/ExecutePipeline/ExecutePipeline.js

@@ -37,6 +37,9 @@ define([
 
         this._currentSave = Q();
         this.changes = {};
+        this.creations = {};
+        this.deletions = [];
+        this.createIdToMetadataId = {};
         this.initRun();
     };
 
@@ -149,7 +152,7 @@ define([
         // before continuing
         this._currentSave = this._currentSave
             .then(() => this.updateForkName(this.pipelineName))
-            .then(() => this.applyChanges())
+            .then(() => this.applyModelChanges())
             .then(() => CreateExecution.prototype.save.call(this, msg))
             .then(result => {
                 var msg;