Udated runtime contract to stop doing unnecessary imports and introduced dedicated process registry. Confirmed working.
This commit is contained in:
+18
-24
@@ -2,7 +2,7 @@ import logging
|
||||
import os
|
||||
|
||||
from core.runtime.stopsignals import checkDependency
|
||||
from core.execution.processregistry import registerProcess
|
||||
from core.execution import processregistry
|
||||
from rich.progress import track
|
||||
from core.console import console
|
||||
|
||||
@@ -10,8 +10,6 @@ def buildRecursive(definition, outputs):
|
||||
process_log = True
|
||||
recursive_log = os.environ.get("LOGS_RECURSIVE", "false").lower() == "true"
|
||||
|
||||
#print(outputs[definition.recursiveset["source"]])
|
||||
|
||||
source = outputs.get(definition.recursiveset["source"])
|
||||
|
||||
if source is None or source == []:
|
||||
@@ -19,7 +17,7 @@ def buildRecursive(definition, outputs):
|
||||
logging.debug(f"No items available in {source} for {definition.name}")
|
||||
return outputs, []
|
||||
|
||||
registerProcess(definition, outputs)
|
||||
currentOp = processregistry.registerProcess(definition)
|
||||
|
||||
responses = []
|
||||
from core.execution.executioncontract import executeContract
|
||||
@@ -32,7 +30,6 @@ def buildRecursive(definition, outputs):
|
||||
reverse = definition.recursiveset.get("reverse", False)
|
||||
|
||||
if reverse:
|
||||
#print("reversing")
|
||||
items = reversed(items)
|
||||
|
||||
for item in track(
|
||||
@@ -46,37 +43,28 @@ def buildRecursive(definition, outputs):
|
||||
|
||||
outputs["stopsignals"] = []
|
||||
for job in recursiveset:
|
||||
processregistry.registeredOp = currentOp
|
||||
|
||||
stop = checkDependency(job, outputs, definition)
|
||||
|
||||
#print(f"STOP SIGNAL DEBUGGING during {job.name}")
|
||||
#print(outputs["stopsignals"])
|
||||
#print(stop)
|
||||
stop = checkDependency(job)
|
||||
|
||||
if stop:
|
||||
#outputs["processes"][definition.name]["stopsignals"].append(job.alias)
|
||||
#print("STOP SIGNAL DETECTED")
|
||||
#print(f"Stopping {job.name} due to stop signal")
|
||||
continue
|
||||
|
||||
#print(f"RECURSIVE DEBUG DURING {job.name}")
|
||||
#print(item)
|
||||
#outputs, response = buildAction(job, outputs)
|
||||
if job.kind == "action":
|
||||
outputs["processes"][definition.name][job.name] = {}
|
||||
currentOp[job.name] = {}
|
||||
|
||||
response = executeContract(job, outputs, definition)
|
||||
response = executeContract(job, outputs)
|
||||
responses.append(response)
|
||||
if job.kind == "action":
|
||||
outputs["processes"][definition.name][job.name]["response"] = response
|
||||
|
||||
outputs["processes"][definition.name]["completed"].append(job.alias)
|
||||
if job.kind == "action":
|
||||
currentOp[job.name]["response"] = response
|
||||
|
||||
currentOp["completed"].append(job.alias)
|
||||
|
||||
if process_log:
|
||||
stopsignal_count = len(outputs["processes"][definition.name]["stopsignals"])
|
||||
completed_count = len(outputs["processes"][definition.name]["completed"])
|
||||
completed_count = len(currentOp["completed"])
|
||||
|
||||
if completed_count == 1 and stopsignal_count == 1:
|
||||
if completed_count == 0:
|
||||
logging.info(f"{definition.name} bypassed")
|
||||
else:
|
||||
logging.info(f"{definition.name} completed")
|
||||
@@ -88,4 +76,10 @@ def buildRecursive(definition, outputs):
|
||||
|
||||
outputs.pop("stopsignals", None)
|
||||
|
||||
process_registry_logs = False
|
||||
|
||||
if process_registry_logs:
|
||||
import pprint
|
||||
pprint.pprint(processregistry.processes)
|
||||
|
||||
return responses
|
||||
|
||||
Reference in New Issue
Block a user