Agent v2 en cours... added PRIORITY commands management

Etienne Pallier
1 parent ef335590
Showing 4 changed files with 145 additions and 53 deletions Show diff stats
CHANGELOG
VERSION
src/core/pyros_django/agent/Agent.py
src/core/pyros_django/common/models.py
+26-10-2022 (EP): v0.6.5.1
+	- (Agent v2 in progress...) - added partial PRIO cmd management
 26-10-2022 (EP): v0.6.5.0
 	- Agent big update : towards Agent v2 async - already compatible with non sequential commands
 25-10-2022 (AKo): v0.6.4.1
-0.6.5.0
 \ No newline at end of file
+0.6.5.1
 \ No newline at end of file
@@ -643,9 +643,14 @@ class Agent:
         # Declaration of Instance attributes, default values
         #self.UP_SINCE = datetime.utcnow()
         self.__UP_SINCE = datetime.now(tz=timezone.utc)
+        self.__ROUTINE_PROCESS_BEFORE_IS_RUNNING:bool = False
+        self.__ROUTINE_PROCESS_AFTER_IS_RUNNING:bool = False
         self.__test_cmd_received_num:int = 0  # only for tests
         # Current Command running
         self.__CC :AgentCmd = None
+        # Previous Command running
+        self.__CC_prev :AgentCmd = None
+        # Current Command exception (if occurs)
         self.__CCE :Exception = None
         self.name = "Generic Agent"
         self.__status :str = None
@@ -1104,7 +1109,7 @@ class Agent:
         self.DO_MAIN_LOOP = True
         self.__CC = None
         self.__CC_prev = None
-        self._CCE = None
+        self.__CCE = None
         while self.DO_MAIN_LOOP:
             # EXIT because of nb of iterations ?
             if nb_iter is not None:
@@ -1155,17 +1160,25 @@ class Agent:
  
         # TRY TO START NEXT COMMAND if exists and if current cmd is finished
         #if self.__CC is None or self.__CC.is_finished() or self.__priority_cmd_received():
-        if not self.__CC or self.__CC.is_finished() or self.__priority_cmd_received():
+        CMD_PRIO=None
+        if not self.__CC or self.__CC.is_finished() or (CMD_PRIO := self.__priority_cmd_received() is not None) :
             print()
             print()
             log.info("*"*10 + " NEXT CMD RECEIVED PROCESSING (START) " + "*"*10 + '\n')
-            if self.__priority_cmd_received(): self.__CC_prev = self.__CC
-            self.__CC = self.__get_next_received_command()
+            ##if self.__priority_cmd_received() and self.__CC.is_running(): self.__CC_prev = self.__CC
+            # Save "current cmd" as now "previous cmd"
+            self.__CC_prev = self.__CC
+            # New CC will be either :
+            # - None if no new command received
+            # or
+            # - a priority command if exists in the commands list, otherwise the next command in the list
+            self.__CC = CMD_PRIO if CMD_PRIO else self.__get_next_received_command()
             if self.__CC:
-                try:
+                try: 
+                    # If CC is a priority cmd, it will be executed SEQUENTIALLY
+                    # Thus this call will be blocking until exec finished
                     self.__start_next_cmd()
-                except Exception as e:
-                    self.__CCE = e
+                except Exception as e: self.__CCE=e
             log.info("*"*10 + " NEXT CMD RECEIVED PROCESSING (END) " + "*"*10 + "\n")
  
         # Check if running cmd timeout
@@ -1173,7 +1186,7 @@ class Agent:
             self.__CC.set_as_exec_timeout()
  
         # If (SEQUENTIAL OR PARALLEL) current cmd is finished => process exception if exists (and test result if in testing mode)
-        if ( self.__CC and self.__CC.is_finished() ) or self.__CCE : 
+        if ( self.__CC and self.__CC.is_finished() ) or self.__CCE :
             self.__process_finished_cmd()
             self.__CC = None
  
@@ -1210,12 +1223,29 @@ class Agent:
         #self.CC = None
  
  
-
-
-
     def __priority_cmd_received(self):
-        return False
-
+        '''
+        Return True if priority cmd found in the commands list
+        (if not irrelevant or running or expired)
+        '''
+        with transaction.atomic():
+            commands = AgentCmd.get_pending_and_running_commands_for_agent(self.name)
+        if not commands.exists(): return None
+        for cmd in commands:
+            #if cmd.name in ("do_exit", "do_abort", "do_flush_commands"): break
+            #if cmd.name in ("do_exit", "do_abort"): break
+            if cmd.is_agent_general_priority_cmd(): 
+                # If prio cmd but irrelevant => skip it
+                if cmd.name in ("do_stop","do_restart") and self.__cmd_was_sent_before_my_start(cmd):
+                    cmd.set_as_skipped("Cmd stop/restart was sent before agent start => skip it")
+                    #return None
+                    continue
+                # if prio cmd already running, wait for it to finish
+                if cmd.is_running():
+                    return None
+                if not cmd.is_expired():
+                    return cmd
+        return None
  
  
     # process self.current_cmd_exception
@@ -1286,6 +1316,23 @@ class Agent:
         #self._DO_EXIT=True
         #exit(0)
  
+        wait_nbsec = 2
+
+        # Waiting for current (in fact previous) command still running if need be (if exists)
+        while self.__CC_prev and self.__CC_prev.is_running():
+            log.info(f"The previous command {self.__CC_prev.name} is still running => Waiting for it to finish (waiting {wait_nbsec} sec...)")
+            self.waitfor(wait_nbsec)
+
+        # Waiting for current ROUTINE BEFORE process to finish if still running
+        while self.__ROUTINE_PROCESS_BEFORE_IS_RUNNING:
+            log.info(f"The ROUTINE BEFORE process is still running => Waiting for it to finish (waiting {wait_nbsec} sec...)")
+            self.waitfor(wait_nbsec)
+
+        # Waiting for current ROUTINE BEFORE process to finish if still running
+        while self.__ROUTINE_PROCESS_AFTER_IS_RUNNING:
+            log.info(f"The ROUTINE AFTER process is still running => Waiting for it to finish (waiting {wait_nbsec} sec...)")
+            self.waitfor(wait_nbsec)
+
         log.info(f"{self.name}: Before exiting, calling do_things_before_exit()")
         self._do_things_before_exit(stopper_agent_name)
         ##self._set_and_log_status(self.AGT_STATUS.EXITING)
@@ -1481,13 +1528,6 @@ class Agent:
  
  
  
-    def __abort_current_running_cmd_if_exists(self):
-        log.info("Aborting current running command if exists...")
-        pass
-    
-    def __abort_current_routine_process_if_exists(self):
-        log.info("Aborting current routine process if exists...")
-        pass
  
  
  
@@ -1506,6 +1546,7 @@ class Agent:
         This is a command or set of processings that this agent does or commands that it sends regularly,
         at each iteration
         """
+        self.__ROUTINE_PROCESS_BEFORE_IS_RUNNING = True
         print()
         print()
         log.info("*"*10+ " ROUTINE BEFORE (START) "+ "*"*10+ '\n')
@@ -1515,6 +1556,7 @@ class Agent:
  
         print()
         log.info("*"*10 + " ROUTINE BEFORE (END) "+ "*"*10)
+        self.__ROUTINE_PROCESS_BEFORE_IS_RUNNING = False
  
  
     def __routine_process_after(self):
@@ -1528,6 +1570,7 @@ class Agent:
         This is a command or set of processings that this agent does or commands that it sends regularly,
         at each iteration
         """
+        self.__ROUTINE_PROCESS_AFTER_IS_RUNNING = True
         print()
         print()
         log.info("*"*10+ " ROUTINE AFTER (START) "+ "*"*10+ '\n')
@@ -1537,6 +1580,7 @@ class Agent:
  
         print()
         log.info("*"*10 + " ROUTINE AFTER (END) "+ "*"*10 + '\n')
+        self.__ROUTINE_PROCESS_AFTER_IS_RUNNING = False
  
  
     # To be overridden by subclasses
@@ -2228,7 +2272,41 @@ class Agent:
         # Default result (null)
         result = None
  
-        # STOP kind of command ?
+        #
+        # - 1) PRIORITY commands
+        #
+
+        # 1.a) not disturbing commands
+
+        if cmd_name == "get_specific_cmds":
+            result = self.get_specific_cmds()
+            '''
+            # CmdUnimplementedException if one specific command (in the list) is unknown
+            try:
+                result = self.get_specific_cmds()
+            except CmdExceptionUnimplemented as e:
+                # raise CmdUnimplementedException("get_specific_cmds.unknown_cmd_name")
+                #cmd.set_as_exec_error("EXCEPTION - One specific cmd is unimplemented: "+e.cmd_name)
+                raise CmdExceptionExecError(cmd, "EXCEPTION - One specific cmd is unimplemented => "+e.cmd_name) from None
+            '''
+
+        #elif cmd_name in ("do_flush_commands"):
+        elif cmd_name == "do_flush_commands":
+            "flush_commands received: Delete all pending commands"
+            self.do_flush_commands()
+            #cmd.set_result('DONE')
+            result = "FLUSH DONE"
+
+        elif cmd_name == "do_exec_commands":
+            pass
+
+        # 1.b) partially disturbing commands
+
+        elif cmd_name == "do_stop_current":
+            what = 'both' if not cmd.args else cmd.args[0]
+            result = self.do_stop_current(what)
+
+        # 1.c) really disturbing commands (STOP command type)
         if cmd_name in ("do_stop", 
                         "do_restart",
                         #@ deprecated
@@ -2255,6 +2333,12 @@ class Agent:
             when = 'asap' if not cmd.args else cmd.args[0]
             result = self.do_stop_or_restart(False if cmd_name=='do_stop' else True,  when)
  
+
+
+        #
+        # - 2) NO priority commands
+        #
+
         elif cmd_name == "get_state":
             result = self.get_state()
  
@@ -2273,13 +2357,6 @@ class Agent:
             result = "MODE is " + mode
             #time.sleep(1)
             #self.waitfor(1)
-        
-        #elif cmd_name in ("do_flush_commands"):
-        elif cmd_name == "do_flush_commands":
-            "flush_commands received: Delete all pending commands"
-            self.do_flush_commands()
-            #cmd.set_result('DONE')
-            result = "FLUSH DONE"
  
         elif cmd_name == "do_eval": 
             #if not cmd_args: raise ValueError()
@@ -2288,17 +2365,6 @@ class Agent:
             #result = eval(cmd_args)
             result = self.do_eval(cmd_args[0])
  
-        elif cmd_name == "get_specific_cmds":
-            result = self.get_specific_cmds()
-            '''
-            # CmdUnimplementedException if one specific command (in the list) is unknown
-            try:
-                result = self.get_specific_cmds()
-            except CmdExceptionUnimplemented as e:
-                # raise CmdUnimplementedException("get_specific_cmds.unknown_cmd_name")
-                #cmd.set_as_exec_error("EXCEPTION - One specific cmd is unimplemented: "+e.cmd_name)
-                raise CmdExceptionExecError(cmd, "EXCEPTION - One specific cmd is unimplemented => "+e.cmd_name) from None
-            '''
  
         cmd.set_as_processed(result)
         log.info("...Agent level GENERAL cmd has been executed")
@@ -2512,17 +2578,35 @@ class Agent:
  
     # Stop currently running cmd or routine
     def do_stop_current(self, what:str):
-        if what == "cmd": self.do_stop_current_cmd()
-        if what == "routine": self.do_stop_current_routine()
+        if what == "cmd": 
+            self.__do_stop_current_cmd_if_exists()
+        if what == "routine": 
+            self.__do_stop_current_routine_before_if_exists()
+            self.__do_stop_current_routine_after_if_exists()
         if what == "both":
-            self.do_stop_current_cmd()
-            self.do_stop_current_routine()
+            self.__do_stop_current_cmd_if_exists()
+            self.__do_stop_current_routine_before_if_exists()
+            self.__do_stop_current_routine_after_if_exists()
         # Bad arg
         raise CmdExceptionBadArgs(self.CC)
-    def do_stop_current_cmd(self):
-        pass
-    def do_stop_current_routine(self):
-        pass
+    #FIXME: TODO
+    def __do_stop_current_cmd_if_exists(self):
+        if self.__CC_prev and self.__CC_prev.is_running():
+            log.info("Aborting current running command if exists...")
+            pass
+
+    #FIXME: TODO
+    def __do_stop_current_routine_before_if_exists(self):
+        if self.__ROUTINE_PROCESS_BEFORE_IS_RUNNING:
+            log.info("Aborting current routine BEFORE process if exists...")
+            pass
+    #FIXME: TODO
+    def __do_stop_current_routine_after_if_exists(self):
+        if self.__ROUTINE_PROCESS_AFTER_IS_RUNNING:
+            log.info("Aborting current routine AFTER process if exists...")
+            pass
+
+
  
     def do_exit(self): self.do_stop("asap");
     def do_abort(self): self.do_stop("now");
@@ -2532,8 +2616,10 @@ class Agent:
     def do_stop_or_restart(self, restart:bool=False, when:str='asap'):
         if when not in ('asap','now','noprio'): raise CmdExceptionBadArgs(self.CC)
         # NOT PRIO
+        '''
         if when == "noprio":
             pass
+        '''
         # log last agent mode & status
         self.__set_and_log_status(self.AGT_STATUS.RESTARTING if restart else self.AGT_STATUS.EXITING)
         #self._log_agent_state()
@@ -626,7 +626,7 @@ class AgentCmd(models.Model):
         # ARGS :
         #   - "cmd" : stop currently running specific command
         #   - "routine" : stop currently running routine process
-        #   - "both" : stop currently running current cmd and routine process
+        #   - "both" (default) : stop currently running current cmd and routine process
         "do_stop_current", # + arg "cmd" or "routine"
         # TODO: a virer => do_stop_current now
         # @deprecated
@@ -649,17 +649,21 @@ class AgentCmd(models.Model):
     # Priority commands, executed as soon as in the received commands list (BEFORE any other pending command)
     # NB : must be a subset of _AGENT_GENERAL_COMMANDS
     _AGENT_GENERAL_PRIORITY_COMMANDS = [
+        
+        # NO STOP commands => will not stop the Agent or any running cmd
         "get_specific_cmds",
         #"do_exit",
         "do_flush_commands", 
         "do_exec_commands"
  
-        "do_restart",
-        # @deprecated
-        "do_restart_loop",
-        
+        # PARTIAL STOP command => will just stop current running cmd (and routine)
         "do_stop_current",
+        
+        # (TOTAL) STOP commands => will stop the Agent (waiting or not for it to finish what it is currently doing)
         "do_stop",
+        "do_restart",
+        # @deprecated
+        ###"do_restart_loop",
     ]
  
     # -------------- Command CLASS (static) METHODS --------------