AtHeartEngineer/SIMoN
1
0
Fork 0
mirror of https://github.com/JHUAPL/SIMoN.git synced 2026-01-08 22:37:56 -05:00
Code Issues Packages Projects Releases Wiki Activity

race condition bug, if one model finishes its increments and publishes outputs before a recipient model receives the increment and clears its validated messages

Browse Source 
Former-commit-id: c9467009df61f2986294d730abd2adc89e1fb028
This commit is contained in:
Michael T. Kelbaugh
2020-03-30 13:52:57 -04:00
parent f6de1fd6f8
commit f879bd1c4b
1 changed files with 27 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
outer_wrapper.py
View File

@@ -180,7 +180,7 @@ class OuterWrapper(ABC):
self.input_schemas = None self.input_schemas = None
self.output_schemas = None self.output_schemas = None
self.validated_schemas = {} self.validated_messages = {"this_incstep": {}, "next_incstep": {}}
self.generic_output_schema = ( self.generic_output_schema = (
"{" "{"
' "type": "object",' ' "type": "object",'
@@ -389,18 +389,20 @@ class OuterWrapper(ABC):
# validate against input schemas # validate against input schemas
if ( if (
incstep > 1 incstep > 1
and len(self.validated_schemas) != self.num_expected_inputs and len(self.validated_messages["this_incstep"])
!= self.num_expected_inputs
): ):
logging.critical( logging.critical(
f"number of validated schemas {len(self.validated_schemas)} != num_expected_inputs {self.num_expected_inputs}" f"number of validated messages {len(self.validated_messages['this_incstep'])} != num_expected_inputs {self.num_expected_inputs}"
) )
event.set() event.set()
raise RuntimeError raise RuntimeError
# call the inner wrapper # call the inner wrapper
schemas = self.validated_schemas.copy() payloads = {}
self.validated_schemas.clear() for schema, message in self.validated_messages["this_incstep"].items():
results = self.increment(**schemas) payloads[schema] = message["payload"]
results = self.increment(**payloads)
# validate against output schemas # validate against output schemas
for schema_name, data_msg in results.items(): for schema_name, data_msg in results.items():
@@ -462,9 +464,21 @@ class OuterWrapper(ABC):
self.status = "ready" self.status = "ready"
elif ( elif (
len(self.validated_schemas) == self.num_expected_inputs len(self.validated_messages["next_incstep"])
== self.num_expected_inputs
): ):
# all input messages have been received and all input schemas have been validated # ready for an increment
self.validated_messages[
"this_incstep"
] = self.validated_messages["next_incstep"].copy()
self.validated_messages["next_incstep"].clear()
self.status = "ready"
elif (
len(self.validated_messages["this_incstep"])
== self.num_expected_inputs
):
# ready for an increment
self.status = "ready" self.status = "ready"
else: else:
@@ -634,10 +648,11 @@ class OuterWrapper(ABC):
logging.info( logging.info(
f"schema {name} validated incoming message from {message['source']}" f"schema {name} validated incoming message from {message['source']}"
) )
if name in self.validated_schemas: if name in self.validated_messages["next_incstep"]:
logging.error( logging.error(
f"schema {name} already validated a message: {message}" f"schema {name} already validated a message: {self.validated_messages['next_incstep'][name]}"
) )
logging.error(f"new message: {message}")
return False return False
else: else:
matched.append(schema) matched.append(schema)
@@ -681,7 +696,8 @@ class OuterWrapper(ABC):
][item]["properties"]["data"].get("unit", "") ][item]["properties"]["data"].get("unit", "")
message["payload"][item]["granularity"] = dest_gran message["payload"][item]["granularity"] = dest_gran
self.validated_schemas[name] = message["payload"] self.validated_messages["next_incstep"][name] = message
except ValidationError: except ValidationError:
logging.debug("validation error") logging.debug("validation error")
except json.JSONDecodeError: except json.JSONDecodeError: