fix:update os.wait4 to os.waitpid

Author: cunla

scheduler/models/task.py

@@ -187,9 +187,9 @@ def is_scheduled(self) -> bool:
             return False
         # check whether job_id is in scheduled/queued/active jobs
         res = (
-                (self.job_name in self.rqueue.scheduled_job_registry.all())
-                or (self.job_name in self.rqueue.queued_job_registry.all())
-                or (self.job_name in self.rqueue.active_job_registry.all())
+            (self.job_name in self.rqueue.scheduled_job_registry.all())
+            or (self.job_name in self.rqueue.queued_job_registry.all())
+            or (self.job_name in self.rqueue.active_job_registry.all())
         )
         # If the job_id is not scheduled/queued/started,
         # update the job_id to None. (The job_id belongs to a previous run which is completed)

scheduler/tests/test_settings.py

@@ -9,23 +9,24 @@
 
 
 class TestWorkerAdmin(SchedulerBaseCase):
-
     def setUp(self):
         from scheduler.settings import SCHEDULER_CONFIG
+
         self.old_settings = SCHEDULER_CONFIG
 
     def tearDown(self):
         from scheduler import settings as scheduler_settings
+
         scheduler_settings.SCHEDULER_CONFIG = self.old_settings
 
     def test_scheduler_config_as_dict(self):
         from scheduler.settings import SCHEDULER_CONFIG
+
         settings.SCHEDULER_CONFIG = dict(
             EXECUTIONS_IN_PAGE=SCHEDULER_CONFIG.EXECUTIONS_IN_PAGE + 1,
             SCHEDULER_INTERVAL=SCHEDULER_CONFIG.SCHEDULER_INTERVAL + 1,
             BROKER=Broker.REDIS,
             CALLBACK_TIMEOUT=SCHEDULER_CONFIG.SCHEDULER_INTERVAL + 1,
-
             DEFAULT_SUCCESS_TTL=SCHEDULER_CONFIG.DEFAULT_SUCCESS_TTL + 1,
             DEFAULT_FAILURE_TTL=SCHEDULER_CONFIG.DEFAULT_FAILURE_TTL + 1,
             DEFAULT_JOB_TTL=SCHEDULER_CONFIG.DEFAULT_JOB_TTL + 1,
@@ -38,18 +39,19 @@ def test_scheduler_config_as_dict(self):
         )
         conf_settings()
         from scheduler.settings import SCHEDULER_CONFIG
+
         for key, value in settings.SCHEDULER_CONFIG.items():
             self.assertEqual(getattr(SCHEDULER_CONFIG, key), value)
 
     def test_scheduler_config_as_data_class(self):
         from scheduler.settings import SCHEDULER_CONFIG
+
         self.assertEqual(SCHEDULER_CONFIG.EXECUTIONS_IN_PAGE, 20)
         settings.SCHEDULER_CONFIG = SchedulerConfiguration(
             EXECUTIONS_IN_PAGE=1,
             SCHEDULER_INTERVAL=60,
             BROKER=Broker.REDIS,
             CALLBACK_TIMEOUT=1111,
-
             DEFAULT_SUCCESS_TTL=1111,
             DEFAULT_FAILURE_TTL=111111,
             DEFAULT_JOB_TTL=1111,
@@ -62,6 +64,7 @@ def test_scheduler_config_as_data_class(self):
         )
         conf_settings()
         from scheduler.settings import SCHEDULER_CONFIG
+
         for key, value in dataclasses.asdict(settings.SCHEDULER_CONFIG).items():
             self.assertEqual(getattr(SCHEDULER_CONFIG, key), value)
 
@@ -71,7 +74,6 @@ def test_scheduler_config_as_dict_bad_param(self):
             SCHEDULER_INTERVAL=60,
             BROKER=Broker.REDIS,
             CALLBACK_TIMEOUT=1111,
-
             DEFAULT_SUCCESS_TTL=1111,
             DEFAULT_FAILURE_TTL=111111,
             DEFAULT_JOB_TTL=1111,
@@ -81,6 +83,6 @@ def test_scheduler_config_as_dict_bad_param(self):
             DEFAULT_MAINTENANCE_TASK_INTERVAL=111,
             DEFAULT_JOB_MONITORING_INTERVAL=1111,
             SCHEDULER_FALLBACK_PERIOD_SECS=1111,
-            BAD_PARAM='bad_value',  # This should raise an error
+            BAD_PARAM="bad_value",  # This should raise an error
         )
         self.assertRaises(ImproperlyConfigured, conf_settings)

scheduler/worker/worker.py

@@ -13,7 +13,6 @@
 from datetime import timedelta
 from enum import Enum
 from random import shuffle
-from resource import struct_rusage
 from types import FrameType
 from typing import List, Optional, Tuple, Any, Iterable
 
@@ -475,14 +474,14 @@ def kill_job_execution_process(self, sig: signal.Signals = SIGKILL):
             else:
                 raise
 
-    def wait_for_job_execution_process(self) -> Tuple[Optional[int], Optional[int], Optional[struct_rusage]]:
+    def wait_for_job_execution_process(self) -> Tuple[Optional[int], Optional[int]]:
         """Waits for the job execution process to complete.
         Uses `0` as argument as to include "any child in the process group of the current process".
         """
-        pid = stat = rusage = None
+        pid = stat = None
         with contextlib.suppress(ChildProcessError):  # ChildProcessError: [Errno 10] No child processes
-            pid, stat, rusage = os.wait4(self._model.job_execution_process_pid, 0)
-        return pid, stat, rusage
+            pid, stat = os.waitpid(self._model.job_execution_process_pid, 0)
+        return pid, stat
 
     def request_force_stop(self, signum: int, frame: Optional[FrameType]):
         """Terminates the application (cold shutdown).
@@ -634,14 +633,14 @@ def monitor_job_execution_process(self, job: JobModel, queue: Queue) -> None:
         :param job: The Job
         :param queue: The Queue
         """
-        retpid = ret_val = rusage = None
+        retpid = ret_val = None
         job.started_at = utcnow()
         while True:
             try:
                 with SCHEDULER_CONFIG.DEATH_PENALTY_CLASS(
                     self.job_monitoring_interval, JobExecutionMonitorTimeoutException
                 ):
-                    retpid, ret_val, rusage = self.wait_for_job_execution_process()
+                    retpid, ret_val = self.wait_for_job_execution_process()
                 break
             except JobExecutionMonitorTimeoutException:
                 # job execution process has not exited yet and is still running. Send a heartbeat to keep the worker alive.