Skip to content

Commit

Permalink
Update GraphProcessor.ts
Browse files Browse the repository at this point in the history
  • Loading branch information
@AryamanAgrawalIronclad
AryamanAgrawalIronclad committed Jan 13, 2025
1 parent d9e7c57 commit abe29cb
Showing 1 changed file with 31 additions and 31 deletions.
62 changes: 31 additions & 31 deletions packages/core/src/model/GraphProcessor.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -248,15 +248,6 @@ export class GraphProcessor {
return this.#running;
}

emitTraceEvent(eventData: string) {
if (this.#includeTrace) {
this.#emitter.emit(
'trace',
eventData,
);
}
}

constructor(project: Project, graphId?: GraphId, registry?: NodeRegistration, includeTrace?: boolean) {
this.#project = project;
const graph = graphId
Expand Down Expand Up @@ -382,6 +373,15 @@ export class GraphProcessor {
});
}

#emitTraceEvent(eventData: string) {
if (this.#includeTrace) {
this.#emitter.emit(
'trace',
eventData,
);
}
}

on = undefined! as Emittery<ProcessEvents>['on'];
off = undefined! as Emittery<ProcessEvents>['off'];
once = undefined! as Emittery<ProcessEvents>['once'];
Expand Down Expand Up @@ -760,7 +760,7 @@ export class GraphProcessor {
for (const node of this.#graph.nodes) {
if (this.#nodeResults.has(node.id)) {

this.emitTraceEvent(`Node ${node.title} has preloaded data`);
this.#emitTraceEvent(`Node ${node.title} has preloaded data`);

await this.#emitter.emit('nodeStart', {
node,
Expand Down Expand Up @@ -883,7 +883,7 @@ export class GraphProcessor {
return;
}

this.emitTraceEvent(`Node ${node.title} has required inputs nodes: ${inputNodes.map((n) => n.title).join(', ')}`);
this.#emitTraceEvent(`Node ${node.title} has required inputs nodes: ${inputNodes.map((n) => n.title).join(', ')}`);

const attachedData = this.#getAttachedDataTo(node);

Expand All @@ -910,7 +910,7 @@ export class GraphProcessor {
inputNodes.map((inputNode) => {
return async () => {

this.emitTraceEvent(`Fetching required data for node ${inputNode.title} (${inputNode.id})`);
this.#emitTraceEvent(`Fetching required data for node ${inputNode.title} (${inputNode.id})`);

await this.#fetchNodeDataAndProcessNode(inputNode);
};
Expand All @@ -925,32 +925,32 @@ export class GraphProcessor {
const builtInNode = node as BuiltInNodes;

if (this.#ignoreNodes.has(node.id)) {
this.emitTraceEvent(`Node ${node.title} is ignored`);
this.#emitTraceEvent(`Node ${node.title} is ignored`);
return;
}

if (this.runToNodeIds) {
const dependencyNodes = this.getDependencyNodesDeep(node.id);

if (this.runToNodeIds.some((runTo) => runTo !== node.id && dependencyNodes.includes(runTo))) {
this.emitTraceEvent(`Node ${node.title} is excluded due to runToNodeIds`);
this.#emitTraceEvent(`Node ${node.title} is excluded due to runToNodeIds`);
return;
}
}

if (this.#currentlyProcessing.has(node.id)) {
this.emitTraceEvent(`Node ${node.title} is already being processed`);
this.#emitTraceEvent(`Node ${node.title} is already being processed`);
return;
}

// For a loop controller, it can run multiple times, otherwise we already processed this node so bail out
if (this.#visitedNodes.has(node.id) && node.type !== 'loopController') {
this.emitTraceEvent(`Node ${node.title} has already been processed`);
this.#emitTraceEvent(`Node ${node.title} has already been processed`);
return;
}

if (this.#erroredNodes.has(node.id)) {
this.emitTraceEvent(`Node ${node.title} has already errored`);
this.#emitTraceEvent(`Node ${node.title} has already errored`);
return;
}

Expand All @@ -959,7 +959,7 @@ export class GraphProcessor {
// Check if all input nodes are free of errors
for (const inputNode of inputNodes) {
if (this.#erroredNodes.has(inputNode.id)) {
this.emitTraceEvent(`Node ${node.title} has errored input node ${inputNode.title}`);
this.#emitTraceEvent(`Node ${node.title} has errored input node ${inputNode.title}`);
return;
}
}
Expand All @@ -972,15 +972,15 @@ export class GraphProcessor {
});

if (!inputsReady) {
this.emitTraceEvent(`Node ${node.title} has required inputs nodes: ${inputNodes.map((n) => n.title).join(', ')}`);
this.#emitTraceEvent(`Node ${node.title} has required inputs nodes: ${inputNodes.map((n) => n.title).join(', ')}`);
return;
}

// Excluded because control flow is still in a loop - difference between "will not execute" and "has not executed yet"
const inputValues = this.#getInputValuesForNode(node);

if (this.#excludedDueToControlFlow(node, inputValues, nanoid() as ProcessId, 'loop-not-broken')) {
this.emitTraceEvent(`Node ${node.title} is excluded due to control flow`);
this.#emitTraceEvent(`Node ${node.title} is excluded due to control flow`);
return;
}

Expand Down Expand Up @@ -1011,7 +1011,7 @@ export class GraphProcessor {
}

if (waitingForInputNode) {
this.emitTraceEvent(`Node ${node.title} is waiting for input node ${waitingForInputNode}`);
this.#emitTraceEvent(`Node ${node.title} is waiting for input node ${waitingForInputNode}`);
return;
}

Expand All @@ -1028,7 +1028,7 @@ export class GraphProcessor {
}

if (attachedData.races?.completed) {
this.emitTraceEvent(`Node ${node.title} is part of a race that was completed`);
this.#emitTraceEvent(`Node ${node.title} is part of a race that was completed`);
return;
}

Expand All @@ -1037,7 +1037,7 @@ export class GraphProcessor {
if (this.slowMode) {
await new Promise((resolve) => setTimeout(resolve, 250));
}
this.emitTraceEvent(`Finished processing node ${node.title} (${node.id})`);
this.#emitTraceEvent(`Finished processing node ${node.title} (${node.id})`);
this.#visitedNodes.add(node.id);
this.#currentlyProcessing.delete(node.id);
this.#remainingNodes.delete(node.id);
Expand All @@ -1055,10 +1055,10 @@ export class GraphProcessor {
this.#excludedDueToControlFlow(node, this.#getInputValuesForNode(node), nanoid() as ProcessId);

if (!didBreak) {
this.emitTraceEvent(`Loop controller ${node.title} did not break, so we're looping again`);
this.#emitTraceEvent(`Loop controller ${node.title} did not break, so we're looping again`);
for (const loopNodeId of attachedData.loopInfo?.nodes ?? []) {
const cycleNode = this.#nodesById[loopNodeId]!;
this.emitTraceEvent(`Clearing cycle node ${cycleNode.title} (${cycleNode.id})`);
this.#emitTraceEvent(`Clearing cycle node ${cycleNode.title} (${cycleNode.id})`);
this.#visitedNodes.delete(cycleNode.id);
this.#currentlyProcessing.delete(cycleNode.id);
this.#remainingNodes.add(cycleNode.id);
Expand All @@ -1076,7 +1076,7 @@ export class GraphProcessor {
for (const [nodeId] of allNodesForRace) {
for (const [key, abortController] of this.#nodeAbortControllers.entries()) {
if (key.startsWith(nodeId)) {
this.emitTraceEvent(`Aborting node ${nodeId} because other race branch won`);
this.#emitTraceEvent(`Aborting node ${nodeId} because other race branch won`);
abortController.abort();
}
}
Expand Down Expand Up @@ -1142,7 +1142,7 @@ export class GraphProcessor {
// Node is finished, check if we can run any more nodes that depend on this one
this.#processingQueue.addAll(
outputNodes.nodes.map((outputNode) => async () => {
this.emitTraceEvent(`Trying to run output node from ${node.title}: ${outputNode.title} (${outputNode.id})`);
this.#emitTraceEvent(`Trying to run output node from ${node.title}: ${outputNode.title} (${outputNode.id})`);

await this.#processNodeIfAllInputsAvailable(outputNode);
}),
Expand Down Expand Up @@ -1391,7 +1391,7 @@ export class GraphProcessor {
#nodeErrored(node: ChartNode, e: unknown, processId: ProcessId) {
const error = getError(e);
this.#emitter.emit('nodeError', { node, error, processId });
this.emitTraceEvent(`Node ${node.title} (${node.id}-${processId}) errored: ${error.stack}`);
this.#emitTraceEvent(`Node ${node.title} (${node.id}-${processId}) errored: ${error.stack}`);
this.#erroredNodes.set(node.id, error.toString());
}

Expand Down Expand Up @@ -1552,7 +1552,7 @@ export class GraphProcessor {
return processor;
},
trace: (message) => {
this.emitTraceEvent(message);
this.#emitTraceEvent(message);
},
abortGraph: (error) => {
this.abort(error === undefined, error);
Expand All @@ -1579,7 +1579,7 @@ export class GraphProcessor {
typeOfExclusion: ControlFlowExcludedDataValue['value'] = undefined,
) {
if (node.disabled) {
this.emitTraceEvent(`Excluding node ${node.title} because it's disabled`);
this.#emitTraceEvent(`Excluding node ${node.title} because it's disabled`);

this.#visitedNodes.add(node.id);
this.#markAsExcluded(node, processId, inputValues, 'disabled');
Expand Down Expand Up @@ -1613,7 +1613,7 @@ export class GraphProcessor {
if (inputIsExcludedValue && !allowedToConsumedExcludedValue) {
if (!isWaitingForLoop) {
if (inputIsExcludedValue) {
this.emitTraceEvent(`Excluding node ${node.title} because of control flow. Input is has excluded value: ${controlFlowExcludedValues[0]?.[0]}`);
this.#emitTraceEvent(`Excluding node ${node.title} because of control flow. Input is has excluded value: ${controlFlowExcludedValues[0]?.[0]}`);
}

this.#visitedNodes.add(node.id);
Expand Down

0 comments on commit abe29cb

Please sign in to comment.