revert wrong simplification

Author: francesconazzaro

cads_broker/dispatcher.py

@@ -368,7 +368,7 @@ def submit_requests(
         queue = sorted(
             candidates,
             key=lambda candidate: self.qos.priority(
-                candidate, session_write, self.internal_scheduler
+                candidate, session_write
             ),
             reverse=True,
         )
@@ -422,7 +422,7 @@ def run(self) -> None:
                 if (rules_hash := get_rules_hash(self.qos.path)) != self.qos.rules_hash:
                     logger.info("reloading qos rules")
                     self.qos.reload_rules(
-                        session=session_read, scheduler=self.internal_scheduler
+                        session=session_read
                     )
                     self.qos.rules_hash = rules_hash
                 self.qos.environment.set_session(session_read)

cads_broker/qos/QoS.py

@@ -65,17 +65,17 @@ def read_rules(self):
         self.rules.dump()
 
     @locked
-    def reload_rules(self, session, scheduler):
+    def reload_rules(self, session):
         """Allow a 'hot' reloading of the rules.
 
         For example, a thread could be monitoring the time stamp of the rules
         file and call this method.
         """
         self.read_rules()
-        self.reconfigure(session=session, scheduler=scheduler)
+        self.reconfigure(session=session)
 
     @locked
-    def reconfigure(self, session, scheduler):
+    def reconfigure(self, session):
         """Reset the status of the QoS.
 
         This method must be called if the rule_set is changed.
@@ -89,27 +89,40 @@ def reconfigure(self, session, scheduler):
         # Re-register the active tasks
         for request in database.get_running_requests(session=session):
             # Recompute the limits
-            for limit in self.limits_for(request, session, scheduler):
+            for limit in self.limits_for(request, session):
                 limit.increment()
 
     @locked
     def can_run(self, request, session, scheduler):
         """Check if a request can run."""
         properties = self._properties(
-            request=request, session=session, scheduler=scheduler
+            request=request, session=session
         )
         limits = []
+        new_limits = []
         for limit in properties.limits:
             if limit.full(request):
                 limits.append(limit)
+                if str(limit.__hash__()) not in [r.uid for r in request.qos_rules]:
+                    new_limits.append(limit)
+        if len(new_limits):
+            scheduler.append(
+                {
+                    "function": database.add_request_qos_status,
+                    "kwargs": {
+                        "request_uid": request.request_uid,
+                        "rules": limits,
+                    },
+                }
+            )
         permissions = []
         for permission in properties.permissions:
             if not permission.evaluate(request):
                 permissions.append(permission)
         return not len(limits) and not len(permissions)
 
     @locked
-    def _properties(self, request, session, scheduler):
+    def _properties(self, request, session):
         """Return the Properties object associated with a request.
 
         If it does not exists it is created.
@@ -148,17 +161,6 @@ def _properties(self, request, session, scheduler):
         if limit is not None:
             properties.limits.append(limit)
 
-        if len(properties.limits) > 0:
-            scheduler.append(
-                {
-                    "function": database.add_request_qos_status,
-                    "kwargs": {
-                        "request_uid": request.request_uid,
-                        "rules": properties.limits,
-                    },
-                }
-            )
-
         # Add priorities and compute starting priority
         priority = 0
         for rule in self.rules.priorities:
@@ -175,38 +177,38 @@ def _properties(self, request, session, scheduler):
         return properties
 
     @locked
-    def priority(self, request, session, scheduler):
+    def priority(self, request, session):
         """Compute the priority of a request."""
         # The priority of a request increases with time
         return (
-            self._properties(request, session, scheduler).starting_priority
+            self._properties(request, session).starting_priority
             + request.age
         )
 
     def dump(self, out=print):
         self.rules.dump(out)
 
     @locked
-    def status(self, requests, session, scheduler, out=print):
+    def status(self, requests, session, out=print):
         out()
         out("===================================================================")
         out("REQUESTS")
         out("===================================================================")
 
         for request in requests:
-            self._status(request, session, scheduler, out)
+            self._status(request, session, out)
 
         out()
         out("===================================================================")
 
-    def _status(self, request, session, scheduler, out):
+    def _status(self, request, session, out):
         out()
         out("===================================================================")
         out("QoS info for:")
         out(request, request.status)
         out("Priority: {}".format(self.priority(request, session)))
         out("Limits rules:")
-        for limit in self.limits_for(request, session, scheduler):
+        for limit in self.limits_for(request, session):
             out(
                 "    {} ({}/{}) {}".format(
                     limit,
@@ -217,36 +219,36 @@ def _status(self, request, session, scheduler, out):
             )
 
         out("Priorities rules:")
-        for priority in self.priorities_for(request, session, scheduler):
+        for priority in self.priorities_for(request, session):
             out("    {}".format(priority))
 
         out("Permissions rules:")
-        for permission in self.permissions_for(request, session, scheduler):
+        for permission in self.permissions_for(request, session):
             out("    {}".format(permission))
 
     @locked
-    def limits_for(self, request, session, scheduler):
+    def limits_for(self, request, session):
         """Return the limit rules that applies to a request.
 
         Ensure that the properties cache is created if needed.
         """
-        return self._properties(request, session, scheduler).limits
+        return self._properties(request, session).limits
 
     @locked
-    def permissions_for(self, request, session, scheduler):
+    def permissions_for(self, request, session):
         """Return the permission rules that applies to a request.
 
         Ensure that the properties cache is created if needed.
         """
-        return self._properties(request, session, scheduler).permissions
+        return self._properties(request, session).permissions
 
     @locked
-    def priorities_for(self, request, session, scheduler):
+    def priorities_for(self, request, session):
         """Return the priority rules that applies to a request.
 
         Ensure that the properties cache is created if needed.
         """
-        return self._properties(request, session, scheduler).priorities
+        return self._properties(request, session).priorities
 
     @locked
     def user_limit(self, request):
@@ -304,7 +306,7 @@ def notify_start_of_request(self, request, session, scheduler):
         its capacity.
         """
         limits_list = []
-        for limit in self.limits_for(request, session, scheduler):
+        for limit in self.limits_for(request, session):
             limit.increment()
             limits_list.append(limit)
         scheduler.append(
@@ -326,7 +328,7 @@ def notify_end_of_request(self, request, session, scheduler):
         sharing the same limits can run.
         """
         limits_list = []
-        for limit in self.limits_for(request, session, scheduler):
+        for limit in self.limits_for(request, session):
             limit.decrement()
             limits_list.append(limit)