Fix batching random durations

prosimos/batch_processing.py

@@ -34,6 +34,13 @@ def _get_operator_symbols_lt(operator_str: str):
 def _is_greater(op: operator):
     return op in [operator.ge, operator.gt]
 
+def whole_seconds(difference: timedelta) -> int:
+    """
+    Seconds in a time difference, rounded down to a whole second, days included.
+    The rule boundaries are whole seconds; timedelta.seconds rounds down too but drops the days.
+    """
+    return difference // timedelta(seconds=1)
+
 class BATCH_TYPE(Enum):
     SEQUENTIAL = 'Sequential'   # one after another
     CONCURRENT = 'Concurrent'   # tasks are in progress simultaneously 
@@ -119,11 +126,16 @@ def is_true(self, element):
             return is_rule_true
 
         elif self.variable1 == "ready_wt":
+            if queue_size == 1 and self.operator in [">", ">="]:
+                # a single waiting case does not fire on the low boundary:
+                # it waits for a second case up to the high boundary, as get_ready_wt counts it
+                return False
+
             last_enabled_datetime = element["enabled_datetimes"][-1]
             curr_enabled_datetime = element["curr_enabled_at"]
             op = _get_operator_symbols_ge(self.operator)
 
-            ready_wt_sec = (curr_enabled_datetime - last_enabled_datetime).seconds
+            ready_wt_sec = whole_seconds(curr_enabled_datetime - last_enabled_datetime)
             is_rule_true = op(ready_wt_sec, self.value2)
 
             if is_rule_true == False:
@@ -466,7 +478,7 @@ def get_ready_wt(self, element):
                 # happens when no new cases will arrive 
                 return en_time_index, _get_enabled_time_for_wt_rule(prev_item, operator.gt, high_boundary)
 
-            diff = (item - prev_item).seconds
+            diff = whole_seconds(item - prev_item)
             is_batch_enabled_low = diff < low_boundary
 
             if is_batch_enabled_low:
@@ -508,14 +520,18 @@ def get_large_wt(self, element):
         prev_item = first_item
         list_len = len(enabled_dt_with_curr_enabled)
         for en_time_index, item in enumerate(enabled_dt_with_curr_enabled[1:], 1):
-            diff = (item - first_item).seconds
+            diff = whole_seconds(item - first_item)
             is_batch_enabled_low = diff > low_boundary
 
             if not is_batch_enabled_low:
                 if en_time_index == list_len - 1:
                     # last item of the evaluation
                     # this will enable the batch in the future
-                    result = en_time_index, _get_enabled_time_for_wt_rule(first_item, operator.ge, low_boundary)
+                    if en_time_index == 1:
+                        # a single waiting case waits for a second one up to the high boundary
+                        result = en_time_index, _get_enabled_time_for_wt_rule(first_item, operator.gt, high_boundary)
+                    else:
+                        result = en_time_index, _get_enabled_time_for_wt_rule(first_item, operator.ge, low_boundary)
                     break
 
                 prev_item = item
@@ -565,6 +581,21 @@ def _get_min_enabled_time_waiting_time(self, case_id_and_enabled_times,
         wt_res = self.get_ready_wt(draft_element) \
             if rule_type == RULE_TYPE.READY_WT else self.get_large_wt(draft_element)
 
+        if wt_res == (0, None):
+            # no new case reaches the batched task anymore, so:
+            # - a single waiting case fires once it has waited past the high boundary;
+            # - several waiting cases fire once the low boundary has passed, counted from
+            #   the last case's arrival for ready_wt (time since the last arrival),
+            #   or from the first case's arrival for large_wt (time the oldest case has waited)
+            low_boundary, high_boundary = self.ready_wt_boundaries \
+                if rule_type == RULE_TYPE.READY_WT else self.large_wt_boundaries
+            enabled_datetimes = draft_element["enabled_datetimes"]
+            if draft_element["size"] == 1:
+                wt_res = 1, _get_enabled_time_for_wt_rule(enabled_datetimes[0], operator.gt, high_boundary)
+            else:
+                reference = enabled_datetimes[-1] if rule_type == RULE_TYPE.READY_WT else enabled_datetimes[0]
+                wt_res = draft_element["size"], _get_enabled_time_for_wt_rule(reference, operator.ge, low_boundary)
+
         if wt_res == None:
             return None
         else:
@@ -913,7 +944,7 @@ def is_invalid_end(self, num_tasks, first_wt, ready_wt):
         Check whether items waiting for batch execution might be satisfied in the future (valid for further processing)
         or they are invalid (one part of the AND rule could not be satisfied in the future at all)
         :param num_tasks: number of tasks waiting for batch execution
-        :param first_wt: waiting time of the first item (current_point_in_time - first_item.enable_time).seconds
+        :param first_wt: waiting time of the first item, whole_seconds(current_point_in_time - first_item.enable_time)
         :param ready_wt: waiting time of the last task in the batch queue
         :return: whether the rule is invalid
         :rtype: boolean

prosimos/control_flow_manager.py

@@ -10,7 +10,7 @@
 from pix_framework.statistics.distribution import DurationDistribution
 
 from prosimos.batch_processing import (BATCH_TYPE, AndFiringRule,
-                                       BatchConfigPerTask)
+                                       BatchConfigPerTask, whole_seconds)
 from prosimos.exceptions import InvalidBpmnModelException
 from prosimos.weekday_helper import CustomDatetimeAndSeconds
 from prosimos.simulation_execution_stats import SimulationExecutionStats
@@ -19,9 +19,9 @@
 seconds_per_unit = {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800}
 
 class BatchInfoForExecution:
-    def __init__(self, all_case_ids, task_batch_info, curr_task_id, batch_spec, start_time_from_rule):
-        self.case_ids = all_case_ids[curr_task_id].copy()
-        self.task_batch_info = task_batch_info[curr_task_id]
+    def __init__(self, case_ids, task_batch_info, batch_spec, start_time_from_rule):
+        self.case_ids = case_ids
+        self.task_batch_info = task_batch_info
         self.batch_spec = batch_spec
         self.start_time_from_rule = start_time_from_rule
         self.batch_id = str(uuid.uuid4())
@@ -400,18 +400,19 @@ def is_batched_task_enabled(self, task_id: str, enabled_at: CustomDatetimeAndSec
 
         firing_rules: List[AndFiringRule] = task_batch_info.firing_rules
 
-        size_count = self.batch_count[task_id] if self.batch_count.get(task_id, None) != None else 0
+        reached = self._reached_batch(task_id, enabled_at)
+        size_count = len(reached)
 
         if not firing_rules.is_batch_size_enough_for_exec(size_count): #size_count < 2:
             # not enough items for batch execution
             return False, None, None
 
-        waiting_time = [ (enabled_at.datetime - v.datetime).total_seconds() for (_, v) in self.batch_waiting_processes[task_id].items() ]
+        waiting_time = [ (enabled_at.datetime - v.datetime).total_seconds() for (_, v) in reached ]
 
         spec = {
             "size": size_count,
             "waiting_times": waiting_time,
-            "enabled_datetimes": [ v.datetime for (_, v) in self.batch_waiting_processes[task_id].items() ],
+            "enabled_datetimes": [ v.datetime for (_, v) in reached ],
             "curr_enabled_at": enabled_at.datetime,
             "is_triggered_by_batch": True, # specify where from we checking the rule. If not triggered by batch - then we move to midnight time
             "is_only_one_batch_return": False
@@ -425,6 +426,16 @@ def is_batched_task_enabled(self, task_id: str, enabled_at: CustomDatetimeAndSec
         else:
             return self.get_batch_size_no_rules_defined(spec, firing_rules, task_batch_info)
 
+    def _reached_batch(self, task_id: str, enabled_at: CustomDatetimeAndSeconds):
+        """
+        Cases waiting for the batched task that have reached it by enabled_at, in the order they reached it.
+        A case is added to the waiting list when its previous task is computed, which happens ahead of time,
+        so the list also holds cases that reach the batched task only later, and it is not ordered by time.
+        """
+        reached = [ (case_id, reached_at) for (case_id, reached_at) in self.batch_waiting_processes[task_id].items()
+                    if reached_at.datetime <= enabled_at.datetime ]
+        return sorted(reached, key=lambda item: item[1].datetime)
+
     def get_batch_size_no_rules_defined(self, spec, firing_rules, task_batch_info):
         """
         In case no rules were discovered, size distribution is being used to define how batches will be formed.
@@ -471,7 +482,8 @@ def get_batch_size_no_rules_defined(self, spec, firing_rules, task_batch_info):
     def get_start_time(self, task_id, last_task_enabled_time) -> tuple([int, int, CustomDatetimeAndSeconds]):
         task_batch_info = self.batch_info.get(task_id, None)
         firing_rules: List[AndFiringRule] = task_batch_info.firing_rules
-        enabled_times = list(self.batch_waiting_processes[task_id].items())
+        # the waiting list is not ordered by the time cases reached the batched task
+        enabled_times = sorted(self.batch_waiting_processes[task_id].items(), key=lambda item: item[1].datetime)
         batch_enabled_time = firing_rules.get_enabled_time(
             enabled_times,
             last_task_enabled_time,
@@ -939,35 +951,23 @@ def move_batch_if_enabled(self, task_id: str, case_id, enabled_time: CustomDatet
         is_enabled, batch_spec, start_time_from_rule = self.is_batched_task_enabled(task_id, enabled_time)
 
         if is_enabled:
-            batch_info = BatchInfoForExecution(
-                self.batch_waiting_processes,
-                self.batch_info,
-                task_id,
-                batch_spec,
-                start_time_from_rule)
+            batch_info = self.create_batch_info_and_clear_from_queue(
+                task_id, batch_spec, start_time_from_rule, self._reached_batch(task_id, enabled_time)
+            )
             enabled_tasks.append((EnabledTask(task_id, batch_info)))
-            self._clear_batch(task_id, batch_spec)
 
 
-    def _clear_batch(self, next_e, batch_spec):
+    def _clear_batch(self, next_e, case_ids):
         """
         When we passed on the information about the batch for the execution,
         clear that data from here to avoid multiple execution
         """
-        for batch_size in batch_spec:
-            if batch_size != None:
-                # remove first batch_size-element since they are being executed
-                # the rest stays in the queue for being enabled for batch execution
-                curr_index = 0
-                for item_key in list(self.batch_waiting_processes[next_e].keys()):
-                    del self.batch_waiting_processes[next_e][item_key]
-                    curr_index = curr_index + 1
+        # remove exactly the cases being executed
+        # the rest stays in the queue for being enabled for batch execution
+        for case_id in case_ids:
+            del self.batch_waiting_processes[next_e][case_id]
 
-                    if curr_index == batch_size:
-                        # all waiting processes regarding the selected batch was removed
-                        break
-
-                self.batch_count[next_e] = self.batch_count[next_e] - batch_size
+        self.batch_count[next_e] = self.batch_count[next_e] - len(case_ids)
 
 
     def increase_task_count(self, task_id, case_id, enabled_time):
@@ -998,7 +998,7 @@ def is_any_batch_enabled(self, started_datetime: CustomDatetimeAndSeconds):
             is_enabled, batch_spec, start_time_from_rule = self.is_batched_task_enabled(task_id, started_datetime)
             if is_enabled:
                 enabled_task_batch[task_id] = self.create_batch_info_and_clear_from_queue(
-                    task_id, batch_spec, start_time_from_rule
+                    task_id, batch_spec, start_time_from_rule, self._reached_batch(task_id, started_datetime)
                 )
 
         return enabled_task_batch
@@ -1042,30 +1042,34 @@ def get_invalid_batches_if_any(self, current_point_of_time):
                 last_task_in_batch_start = waiting_tasks[last_added_key].datetime
                 start_time_from_rule = max(current_point_of_time.datetime, last_task_in_batch_start)
                 enabled_task_batch[task_id] = self.create_batch_info_and_clear_from_queue(
-                    task_id, batch_spec, start_time_from_rule
+                    task_id, batch_spec, start_time_from_rule, list(waiting_tasks.items())
                 )
                 continue
 
         return enabled_task_batch
 
-    def create_batch_info_and_clear_from_queue(self, task_id: str, batch_spec, start_time_from_rule):
+    def create_batch_info_and_clear_from_queue(self, task_id: str, batch_spec, start_time_from_rule, candidates):
+        """
+        :param candidates: (case_id, enabled_time) pairs the firing rule looked at, in the same order.
+            The batch takes the first of them, as many as batch_spec counts in total
+        """
+        case_ids = dict(candidates[:sum(batch_spec)])
         batch_info = BatchInfoForExecution(
-            self.batch_waiting_processes,
-            self.batch_info,
-            task_id,
+            case_ids,
+            self.batch_info[task_id],
             batch_spec,
             start_time_from_rule)
 
-        self._clear_batch(task_id, batch_spec)
+        self._clear_batch(task_id, case_ids)
         return batch_info
 
     def is_or_rule_invalid(self, waiting_tasks, task_id: str, num_tasks_wait_batch: int, current_point_of_time: CustomDatetimeAndSeconds):
         all_keys = list(waiting_tasks.keys())
 
         first_key = all_keys[0]
-        first_wt = (current_point_of_time.datetime - waiting_tasks[first_key].datetime).seconds
+        first_wt = whole_seconds(current_point_of_time.datetime - waiting_tasks[first_key].datetime)
 
         last_key = all_keys[-1]
-        last_wt = (current_point_of_time.datetime - waiting_tasks[last_key].datetime).seconds
+        last_wt = whole_seconds(current_point_of_time.datetime - waiting_tasks[last_key].datetime)
 
         return self.batch_info[task_id].firing_rules.is_invalid_end(num_tasks_wait_batch, first_wt, last_wt)