while (jobItemItr.hasNext()) {
assert jobResultItr.hasNext();
final CalculationJobItem jobItem = jobItemItr.next();
final CalculationJobResultItem jobResultItem = jobResultItr.next();
// Mark the node that corresponds to this item
final DependencyNode node = graph.getNodeProducing(jobItem.getOutputs()[0]);
if (jobResultItem.isFailed()) {
getCycle().markFailed(node);
} else {
getCycle().markExecuted(node);
}
// Process the streamed result fragment
final ExecutionLogMode executionLogMode = logModes.getLogMode(node);
final AggregatedExecutionLog aggregatedExecutionLog;
if (executionLogMode == ExecutionLogMode.FULL) {
final ExecutionLog log = jobResultItem.getExecutionLog();
MutableExecutionLog logCopy = null;
final Set<AggregatedExecutionLog> inputLogs = new LinkedHashSet<AggregatedExecutionLog>();
Set<ValueSpecification> missing = jobResultItem.getMissingInputs();
if (!missing.isEmpty()) {
if (logCopy == null) {
logCopy = new MutableExecutionLog(log);
}
logCopy.add(new SimpleLogEvent(log.hasException() ? LogLevel.WARN : LogLevel.INFO, toString("Missing input", missing)));
}
missing = jobResultItem.getMissingOutputs();
if (!missing.isEmpty()) {
if (logCopy == null) {
logCopy = new MutableExecutionLog(log);
}
logCopy.add(new SimpleLogEvent(LogLevel.WARN, toString("Failed to produce output", missing)));
}
for (final ValueSpecification inputValueSpec : node.getInputValues()) {
final DependencyNodeJobExecutionResult nodeResult = jobExecutionResultCache.get(inputValueSpec);
if (nodeResult == null) {
// Market data
continue;
}
inputLogs.add(nodeResult.getAggregatedExecutionLog());
}
aggregatedExecutionLog = DefaultAggregatedExecutionLog.fullLogMode(node, (logCopy != null) ? logCopy : log, inputLogs);
} else {
EnumSet<LogLevel> logs = jobResultItem.getExecutionLog().getLogLevels();
boolean copied = false;
for (final ValueSpecification inputValueSpec : node.getInputValues()) {
final DependencyNodeJobExecutionResult nodeResult = jobExecutionResultCache.get(inputValueSpec);
if (nodeResult == null) {
// Market data
continue;
}
if (logs.containsAll(nodeResult.getAggregatedExecutionLog().getLogLevels())) {
continue;
}
if (!copied) {
copied = true;
logs = EnumSet.copyOf(logs);
}
logs.addAll(nodeResult.getAggregatedExecutionLog().getLogLevels());
}
aggregatedExecutionLog = DefaultAggregatedExecutionLog.indicatorLogMode(logs);
}
final DependencyNodeJobExecutionResult jobExecutionResult = new DependencyNodeJobExecutionResult(computeNodeId, jobResultItem, aggregatedExecutionLog);
node.gatherTerminalOutputValues(terminalOutputs);
for (ValueSpecification output : node.getOutputValues()) {
jobExecutionResultCache.put(output, jobExecutionResult);
}
}
_issueFragmentResults |= !terminalOutputs.isEmpty();
}