Skip to content

Commit 09c28f7

Browse files
brollbbrollb
authored
Moved 2PC logic to TwoPhaseCommit plugin. Closes #1233 (#1306)
* Moved 2PC logic to TwoPhaseCommit plugin. Closes #1233 * Update CreatedNodes in node caches * WIP Add tests for setPointer * WIP delete nodes using CreatedNode object * Added loadChildren to 2PC * WIP Updatedd ExecuteJob.Metadata.js * WIP added working edits, debug statements, comments * WIP Wrap the core instead of adding methods to plugin * WIP updated plugins to use wrapped core * WIP removed extra console logs, debug comments * WIP removed useless comments * WIP Update ExecutePipeline tests * WIP Update model editing to use (wrapped) core
1 parent 20b0910 commit 09c28f7

File tree

15 files changed

+1347
-747
lines changed

15 files changed

+1347
-747
lines changed

‎src/common/plugin/LocalExecutor.js‎

Lines changed: 14 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ define([
1616
.find(cntr => this.isMetaTypeOf(cntr, this.META.Outputs));
1717

1818
const dataNodes = await this.core.loadChildren(outputContainer);
19-
const dataInfo = this.getAttribute(dataNodes[0], 'data');
19+
const dataInfo = this.core.getAttribute(dataNodes[0], 'data');
2020

2121
// Pass the dataInfo to the next nodes
2222
const outputs = (await this.getOutputs(node))
@@ -39,7 +39,7 @@ define([
3939
});
4040

4141
const saveDir = containers.find(c =>
42-
this.getAttribute(c, 'name').toLowerCase().includes('artifacts')
42+
this.core.getAttribute(c, 'name').toLowerCase().includes('artifacts')
4343
) || containers[0];
4444

4545
return saveDir || this.rootNode; // default to rootNode
@@ -50,8 +50,8 @@ define([
5050
const artifacts = await this.core.loadChildren(artifactsDir);
5151
const currNameHashPairs = artifacts
5252
.map(node => [
53-
this.getAttribute(node, 'name'),
54-
this.getAttribute(node, 'data')
53+
this.core.getAttribute(node, 'name'),
54+
this.core.getAttribute(node, 'data')
5555
]);
5656
const inputs = await this.getInputs(node);
5757
const ids = inputs.map(i => this.core.getPath(i[2]));
@@ -63,28 +63,29 @@ define([
6363

6464
// Remove nodes that already exist
6565
const dataNodes = incomingData.filter(dataNode => {
66-
const hash = this.getAttribute(dataNode, 'data');
66+
const hash = this.core.getAttribute(dataNode, 'data');
6767
const name = this.core.getOwnAttribute(node, 'saveName') ||
68-
this.getAttribute(dataNode, 'name');
68+
this.core.getAttribute(dataNode, 'name');
6969

7070
return !(currNameHashPairs
7171
.find(pair => pair[0] === name && pair[1] === hash));
7272
});
7373

7474
const saveDir = `${this.projectId}/artifacts/`;
7575
const storage = await this.getStorageClient();
76+
const createParams = {base: this.META.Data, parent: artifactsDir};
7677
for (let i = dataNodes.length; i--;) {
77-
const artifact = this.createNode('Data', artifactsDir);
78+
const artifact = this.core.createNode(createParams);
7879
const name = this.core.getOwnAttribute(node, 'saveName') ||
79-
this.getAttribute(dataNodes[i], 'name');
80+
this.core.getAttribute(dataNodes[i], 'name');
8081
const createdAt = Date.now();
81-
const originalData = JSON.parse(this.getAttribute(dataNodes[i], 'data'));
82+
const originalData = JSON.parse(this.core.getAttribute(dataNodes[i], 'data'));
8283
const userAsset = await storage.copy(originalData, saveDir + name);
8384

84-
this.setAttribute(artifact, 'data', JSON.stringify(userAsset));
85-
this.setAttribute(artifact, 'name', name);
86-
this.setAttribute(artifact, 'createdAt', createdAt);
87-
this.setPointer(artifact, 'origin', inputs[0][2]);
85+
this.core.setAttribute(artifact, 'data', JSON.stringify(userAsset));
86+
this.core.setAttribute(artifact, 'name', name);
87+
this.core.setAttribute(artifact, 'createdAt', createdAt);
88+
this.core.setPointer(artifact, 'origin', inputs[0][2]);
8889
}
8990

9091
this.logger.info(`Saved ${dataNodes.length} artifacts in ${this.projectId}.`);

‎src/common/plugin/Operation.js‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ define([
1010
};
1111

1212
OperationOps.prototype.getOutputs = function (node) {
13-
const code = this.getAttribute(node, 'code');
13+
const code = this.core.getAttribute(node, 'code');
1414
let outputNames = [];
1515
if (code) {
1616
const operation = OperationCode.findOperation(code);
@@ -25,7 +25,7 @@ define([
2525
};
2626

2727
OperationOps.prototype.getInputs = function (node) {
28-
const code = this.getAttribute(node, 'code');
28+
const code = this.core.getAttribute(node, 'code');
2929
let inputNames = [];
3030
if (code) {
3131
const operation = OperationCode.findOperation(code);
@@ -50,8 +50,8 @@ define([
5050
var bases = outputs.map(node => this.core.getMetaType(node));
5151
// return [[arg1, Type1, node1], [arg2, Type2, node2]]
5252
return outputs.map((node, i) => [
53-
this.getAttribute(node, 'name'),
54-
this.getAttribute(bases[i], 'name'),
53+
this.core.getAttribute(node, 'name'),
54+
this.core.getAttribute(bases[i], 'name'),
5555
node
5656
]);
5757
});

‎src/plugins/ExecuteJob/ExecuteJob.Metadata.js‎

Lines changed: 39 additions & 72 deletions
Original file line numberDiff line numberDiff line change
@@ -10,11 +10,10 @@ define([
1010
this._markForDeletion = {}; // id -> node
1111
this._oldMetadataByName = {}; // name -> id
1212
this.createdMetadataIds = {};
13-
14-
this.plotLines = {};
13+
this.createIdToMetadataId = {};
1514
};
1615

17-
// I think I should convert these to just a single 'update graph' command
16+
// TODO: Add tests
1817
ExecuteJob.prototype[CONSTANTS.PLOT_UPDATE] = async function (job, state) {
1918
const jobId = this.core.getPath(job);
2019

@@ -23,86 +22,53 @@ define([
2322
let id = jobId + '/' + state.id;
2423
let graph = this.getExistingMetadataById(job, 'Graph', id);
2524
if (!graph) {
26-
graph = this.createNode('Graph', job);
27-
this.setAttribute(graph, 'id', id);
28-
29-
this.createIdToMetadataId[graph] = id;
25+
graph = this.core.createNode({
26+
base: this.META.Graph,
27+
parent: job
28+
});
29+
this.core.setAttribute(graph, 'id', id);
30+
this._metadata[id] = graph;
3031
}
3132

3233
// Apply whatever updates are needed
3334
// Set the plot title
3435
// Only support a single axes for now
3536
const axes = state.axes[0];
36-
this.setAttribute(graph, 'name', axes.title);
37-
this.setAttribute(graph, 'xlabel', axes.xlabel);
38-
this.setAttribute(graph, 'ylabel', axes.ylabel);
37+
this.core.setAttribute(graph, 'name', axes.title);
38+
this.core.setAttribute(graph, 'xlabel', axes.xlabel);
39+
this.core.setAttribute(graph, 'ylabel', axes.ylabel);
3940
this.logger.info(`Updating graph named ${axes.title}`);
4041

41-
// Delete current line nodes
42-
if (!this.isCreateId(graph)) {
43-
//const children = await this.core.loadChildren(graph);
44-
const childIds = this.core.getChildrenPaths(graph);
45-
childIds.forEach(id => this.deleteNode(id));
46-
}
47-
48-
if (this.plotLines[id]) {
49-
this.plotLines[id].forEach(lineId => {
50-
if (this._metadata[lineId]) {
51-
const nodeId = this.core.getPath(this._metadata[lineId]);
52-
this.deleteNode(nodeId);
53-
} else {
54-
const createId = Object.keys(this.createIdToMetadataId)
55-
.find(createId => this.createIdToMetadataId[createId] === lineId);
56-
57-
if (createId) {
58-
this.deleteNode(createId);
59-
}
60-
}
61-
});
62-
}
63-
this.plotLines[id] = [];
42+
const children = await this.core.loadChildren(graph);
43+
children.forEach(node => this.core.deleteNode(node));
6444

6545
// Update the points for each of the lines
6646
axes.lines.forEach((line, index) => {
67-
let lineId = id + '/' + index;
68-
let node = this.createNode('Line', graph);
69-
this.plotLines[id].push(lineId);
70-
this.createIdToMetadataId[node] = lineId;
47+
let node = this.core.createNode({
48+
base: this.META.Line,
49+
parent: graph,
50+
});
7151

72-
this.setAttribute(node, 'name', line.label || `line ${index+1}`);
52+
this.core.setAttribute(node, 'name', line.label || `line ${index+1}`);
7353
let points = line.points.map(pts => pts.join(',')).join(';');
74-
this.setAttribute(node, 'points', points);
54+
this.core.setAttribute(node, 'points', points);
7555
});
7656
};
7757

78-
ExecuteJob.prototype[CONSTANTS.GRAPH_CREATE_LINE] = function (job, graphId, id) {
79-
var jobId = this.core.getPath(job),
80-
graph = this._metadata[jobId + '/' + graphId],
81-
name = Array.prototype.slice.call(arguments, 3).join(' '),
82-
line;
83-
84-
// Create a 'line' node in the given Graph metadata node
85-
name = name.replace(/\s+$/, '');
86-
line = this.createNode('Line', graph);
87-
this.setAttribute(line, 'name', name);
88-
this._metadata[jobId + '/' + id] = line;
89-
this.createIdToMetadataId[line] = jobId + '/' + id;
90-
};
91-
9258
ExecuteJob.prototype[CONSTANTS.IMAGE.BASIC] =
9359
ExecuteJob.prototype[CONSTANTS.IMAGE.UPDATE] =
9460
ExecuteJob.prototype[CONSTANTS.IMAGE.CREATE] = function (job, hash, imgId) {
9561
var name = Array.prototype.slice.call(arguments, 3).join(' '),
9662
imageNode = this._getImageNode(job, imgId, name);
9763

98-
this.setAttribute(imageNode, 'data', hash);
64+
this.core.setAttribute(imageNode, 'data', hash);
9965
};
10066

10167
ExecuteJob.prototype[CONSTANTS.IMAGE.NAME] = function (job, imgId) {
10268
var name = Array.prototype.slice.call(arguments, 2).join(' '),
10369
imageNode = this._getImageNode(job, imgId, name);
10470

105-
this.setAttribute(imageNode, 'name', name);
71+
this.core.setAttribute(imageNode, 'name', name);
10672
};
10773

10874
ExecuteJob.prototype._getImageNode = function (job, imgId, name) {
@@ -116,9 +82,11 @@ define([
11682
imageNode = this._getExistingMetadata(jobId, 'Image', name);
11783
if (!imageNode) {
11884
this.logger.info(`Creating image ${id} named ${name}`);
119-
imageNode = this.createNode('Image', job);
120-
this.setAttribute(imageNode, 'name', name);
121-
this.createIdToMetadataId[imageNode] = id;
85+
imageNode = this.core.createNode({
86+
base: this.META.Image,
87+
parent: job,
88+
});
89+
this.core.setAttribute(imageNode, 'name', name);
12290
}
12391
this._metadata[id] = imageNode;
12492
}
@@ -147,7 +115,7 @@ define([
147115
if (this.isMetaTypeOf(child, this.META.Metadata)) {
148116
id = this.core.getPath(child);
149117
base = this.core.getBase(child);
150-
type = this.getAttribute(base, 'name');
118+
type = this.core.getAttribute(base, 'name');
151119

152120
this._markForDeletion[nodeId][id] = child;
153121
// namespace by metadata type
@@ -166,7 +134,7 @@ define([
166134
// make the deletion ids relative to the job node
167135
this.logger.debug(`About to delete ${idsToDelete.length}: ${idsToDelete.join(', ')}`);
168136
for (i = idsToDelete.length; i--;) {
169-
this.deleteNode(idsToDelete[i]);
137+
this.core.deleteNode(idsToDelete[i]);
170138
}
171139
});
172140
};
@@ -182,14 +150,14 @@ define([
182150
this.logger.debug(`About to delete ${nodeIds.length}: ${nodeIds.join(', ')}`);
183151
for (var i = nodeIds.length; i--;) {
184152
const node = this._markForDeletion[nodeId][nodeIds[i]];
185-
this.deleteNode(this.core.getPath(node));
153+
this.core.deleteNode(this.core.getPath(node));
186154
}
187155
delete this.lastAppliedCmd[nodeId];
188156
delete this.createdMetadataIds[nodeId];
189157
delete this._markForDeletion[nodeId];
190158
}
191159

192-
this.delAttribute(job, 'jobInfo');
160+
this.core.delAttribute(job, 'jobInfo');
193161
};
194162

195163
ExecuteJob.prototype.resultMsg = function(msg) {
@@ -198,21 +166,14 @@ define([
198166
};
199167

200168
ExecuteJob.prototype.getExistingMetadataById = function (job, type, id) {
201-
const createId = Object.keys(this.createIdToMetadataId)
202-
.find(createId => this.createIdToMetadataId[createId] === id);
203-
204-
if (createId) { // on the queue to be created
205-
return createId;
206-
}
207-
208-
if (this._metadata[id]) { // already created
169+
if (this._metadata[id]) {
209170
return this._metadata[id];
210171
}
211172

212-
return this._getExistingMetadata( // exists from prev run
173+
return this._getExistingMetadata( // exists from prev run
213174
this.core.getPath(job),
214175
type,
215-
node => this.getAttribute(node, 'id') === id
176+
node => this.core.getAttribute(node, 'id') === id
216177
);
217178
};
218179

@@ -271,5 +232,11 @@ define([
271232
};
272233
};
273234

235+
ExecuteJob.prototype.onNodeCreated = async function (tmpId, node) {
236+
const id = this.createIdToMetadataId[tmpId];
237+
delete this.createIdToMetadataId[tmpId];
238+
this._metadata[id] = node;
239+
};
240+
274241
return ExecuteJob;
275242
});

0 commit comments

Comments
 (0)