We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Better handling of k8s SIGTERM when stopping workers
We've observed recently that worker pods can take more than 1h to get killed by kubernetes (see internal [slack thread](https://huggingface.slack.com/archives/C04L6P8KNQ5/p1697641144036899) for example).
According to the k8s docs, at one point it sends a TERM and then after a grace period a SIGKILL on all the running processes.
If this keeps happening we need to improve the way we handle those to kill the pods faster.

One idea is to catch SIGTERM in the worker executor (main python process of workers) that would stop the worker loop (its subprocess where the actual code runs).

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `services/worker/src/worker/executor.py`
Content:
```
1 # SPDX-License-Identifier: Apache-2.0
2 # Copyright 2022 The HuggingFace Authors.
3 import asyncio
4 import logging
5 import os
6 import sys
7 from collections.abc import Callable
8 from datetime import datetime, timedelta
9 from random import random
10 from typing import Any, Optional, Union
11 
12 import orjson
13 from filelock import FileLock
14 from libcommon.processing_graph import ProcessingGraph
15 from libcommon.queue import Queue
16 from libcommon.utils import get_datetime
17 from mirakuru import OutputExecutor, ProcessExitedWithError
18 
19 from worker import start_worker_loop
20 from worker.config import AppConfig
21 from worker.job_manager import JobManager
22 from worker.job_runner_factory import JobRunnerFactory
23 from worker.loop import WorkerState
24 
25 START_WORKER_LOOP_PATH = start_worker_loop.__file__
26 
27 
28 async def every(
29     func: Callable[..., Optional[Any]],
30     *args: Any,
31     seconds: Union[float, tuple[float, float]],
32     stop_on: Optional[Any] = None,
33     **kwargs: Any,
34 ) -> None:
35     while True:
36         out = func(*args, **kwargs)
37         if stop_on is not None and out == stop_on:
38             break
39         delay = (
40             seconds[0] + (seconds[1] - seconds[0]) * random() if isinstance(seconds, tuple) else seconds  # nosec B311
41         )
42         await asyncio.sleep(delay)
43 
44 
45 class BadWorkerState(RuntimeError):
46     """Raised when the worker state from the worker read by the executor is not valid."""
47 
48     pass
49 
50 
51 class WorkerExecutor:
52     def __init__(self, app_config: AppConfig, job_runner_factory: JobRunnerFactory, state_file_path: str) -> None:
53         self.app_config = app_config
54         self.job_runner_factory = job_runner_factory
55         self.state_file_path = state_file_path
56         self.processing_graph = ProcessingGraph(self.app_config.processing_graph)
57 
58         max_missing_heartbeats = self.app_config.worker.max_missing_heartbeats
59         heartbeat_interval_seconds = self.app_config.worker.heartbeat_interval_seconds
60         self.max_seconds_without_heartbeat_for_zombies = heartbeat_interval_seconds * max_missing_heartbeats
61 
62         self.heartbeat_interval_seconds = self.app_config.worker.heartbeat_interval_seconds
63         self.max_job_duration_seconds = self.app_config.worker.max_job_duration_seconds
64         self.kill_zombies_interval_seconds = self.app_config.worker.kill_zombies_interval_seconds
65         self.kill_long_job_interval_seconds = self.app_config.worker.kill_long_job_interval_seconds
66 
67     def _create_worker_loop_executor(self) -> OutputExecutor:
68         banner = self.state_file_path
69         start_worker_loop_command = [
70             sys.executable,
71             START_WORKER_LOOP_PATH,
72             "--print-worker-state-path",
73         ]
74         return OutputExecutor(start_worker_loop_command, banner, timeout=10)
75 
76     def start(self) -> None:
77         exceptions = []
78         worker_loop_executor = self._create_worker_loop_executor()
79         worker_loop_executor.start()  # blocking until the banner is printed
80 
81         def custom_exception_handler(loop: asyncio.AbstractEventLoop, context: dict[str, Any]) -> None:
82             nonlocal exceptions
83             # first, handle with default handler
84             loop.default_exception_handler(context)
85 
86             exception = context.get("exception")
87             if exception:
88                 exceptions.append(repr(exception))
89                 loop.stop()
90 
91         loop = asyncio.get_event_loop()
92         loop.set_exception_handler(custom_exception_handler)
93         logging.info("Starting heartbeat.")
94         loop.create_task(every(self.heartbeat, seconds=self.heartbeat_interval_seconds))
95         loop.create_task(
96             every(
97                 self.kill_zombies,
98                 seconds=(
99                     self.kill_zombies_interval_seconds * 0.5,
100                     self.kill_zombies_interval_seconds * 1.5,
101                 ),
102             )
103         )
104         loop.create_task(
105             every(
106                 self.kill_long_job,
107                 worker_loop_executor=worker_loop_executor,
108                 seconds=(
109                     self.kill_long_job_interval_seconds * 0.5,
110                     self.kill_long_job_interval_seconds * 1.5,
111                 ),
112             )
113         )
114         loop.run_until_complete(
115             every(self.is_worker_alive, worker_loop_executor=worker_loop_executor, seconds=1.0, stop_on=False)
116         )
117         if exceptions:
118             raise RuntimeError(f"Some async tasks failed: {exceptions}")
119 
120     def get_state(self) -> Optional[WorkerState]:
121         worker_state_file_path = self.state_file_path
122         if not os.path.exists(worker_state_file_path):
123             return None
124         with FileLock(f"{worker_state_file_path}.lock"):
125             try:
126                 with open(worker_state_file_path, "rb") as worker_state_f:
127                     worker_state = orjson.loads(worker_state_f.read())
128                     return WorkerState(
129                         current_job_info=worker_state.get("current_job_info"),
130                         last_updated=datetime.fromisoformat(worker_state["last_updated"]),
131                     )
132             except (orjson.JSONDecodeError, KeyError) as err:
133                 raise BadWorkerState(f"Failed to read worker state at {worker_state_file_path}") from err
134 
135     def heartbeat(self) -> None:
136         worker_state = self.get_state()
137         if worker_state and worker_state["current_job_info"]:
138             Queue().heartbeat(job_id=worker_state["current_job_info"]["job_id"])
139 
140     def kill_zombies(self) -> None:
141         queue = Queue()
142         zombies = queue.get_zombies(max_seconds_without_heartbeat=self.max_seconds_without_heartbeat_for_zombies)
143         message = "Job manager crashed while running this job (missing heartbeats)."
144         for zombie in zombies:
145             job_runner = self.job_runner_factory.create_job_runner(zombie)
146             job_manager = JobManager(
147                 job_info=zombie,
148                 app_config=self.app_config,
149                 job_runner=job_runner,
150                 processing_graph=self.processing_graph,
151             )
152             job_manager.set_crashed(message=message)
153             logging.info(f"Killing zombie. Job info = {zombie}")
154 
155     def kill_long_job(self, worker_loop_executor: OutputExecutor) -> None:
156         worker_state = self.get_state()
157         if worker_state and worker_state["current_job_info"]:
158             long_job = worker_state["current_job_info"]
159             last_updated = worker_state["last_updated"]
160             coefficient = 10 if long_job["params"]["dataset"] == "cerebras/SlimPajama-627B" else 1
161             if last_updated + timedelta(seconds=coefficient * self.max_job_duration_seconds) <= get_datetime():
162                 _duration_seconds = int((get_datetime() - last_updated).total_seconds())
163                 logging.warning(
164                     f"Job {long_job} exceeded maximum duration of"
165                     f" {self.max_job_duration_seconds} seconds ({_duration_seconds} seconds)."
166                 )
167                 try:
168                     worker_loop_executor.stop()  # raises an error if the worker returned exit code 1
169                 finally:
170                     logging.info(f"Killing a long job. Job info = {long_job}")
171                     job_runner = self.job_runner_factory.create_job_runner(long_job)
172                     job_manager = JobManager(
173                         job_info=long_job,
174                         app_config=self.app_config,
175                         job_runner=job_runner,
176                         processing_graph=self.processing_graph,
177                     )
178                     message = "Job manager was killed while running this job (job exceeded maximum duration)."
179                     job_manager.set_exceeded_maximum_duration(message=message)
180 
181     def is_worker_alive(self, worker_loop_executor: OutputExecutor) -> bool:
182         if worker_loop_executor.running():
183             return True
184         try:
185             worker_loop_executor.stop()  # raises an error if the worker returned unexpected exit code
186         except ProcessExitedWithError as err:
187             explanation = f"exit code f{err.exit_code}"
188             if err.exit_code == -9:
189                 explanation += " SIGKILL - surely an OOM"
190             error_msg = f"Worker crashed ({explanation})"
191             state = self.get_state()
192             if state and state["current_job_info"]:
193                 error_msg += f"when running job_id={state['current_job_info']['job_id']}"
194             logging.error(error_msg)
195             raise
196         return False
197
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/services/worker/src/worker/executor.py b/services/worker/src/worker/executor.py
--- a/services/worker/src/worker/executor.py
+++ b/services/worker/src/worker/executor.py
@@ -3,6 +3,7 @@
 import asyncio
 import logging
 import os
+import signal
 import sys
 from collections.abc import Callable
 from datetime import datetime, timedelta
@@ -89,6 +90,7 @@
                 loop.stop()
 
         loop = asyncio.get_event_loop()
+        loop.add_signal_handler(signal.SIGTERM, self.stop, worker_loop_executor)
         loop.set_exception_handler(custom_exception_handler)
         logging.info("Starting heartbeat.")
         loop.create_task(every(self.heartbeat, seconds=self.heartbeat_interval_seconds))
@@ -117,6 +119,9 @@
         if exceptions:
             raise RuntimeError(f"Some async tasks failed: {exceptions}")
 
+    def stop(self, worker_loop_executor: OutputExecutor) -> None:
+        worker_loop_executor.stop()
+
     def get_state(self) -> Optional[WorkerState]:
         worker_state_file_path = self.state_file_path
         if not os.path.exists(worker_state_file_path):

{"golden_diff": "diff --git a/services/worker/src/worker/executor.py b/services/worker/src/worker/executor.py\n--- a/services/worker/src/worker/executor.py\n+++ b/services/worker/src/worker/executor.py\n@@ -3,6 +3,7 @@\n import asyncio\n import logging\n import os\n+import signal\n import sys\n from collections.abc import Callable\n from datetime import datetime, timedelta\n@@ -89,6 +90,7 @@\n                 loop.stop()\n \n         loop = asyncio.get_event_loop()\n+        loop.add_signal_handler(signal.SIGTERM, self.stop, worker_loop_executor)\n         loop.set_exception_handler(custom_exception_handler)\n         logging.info(\"Starting heartbeat.\")\n         loop.create_task(every(self.heartbeat, seconds=self.heartbeat_interval_seconds))\n@@ -117,6 +119,9 @@\n         if exceptions:\n             raise RuntimeError(f\"Some async tasks failed: {exceptions}\")\n \n+    def stop(self, worker_loop_executor: OutputExecutor) -> None:\n+        worker_loop_executor.stop()\n+\n     def get_state(self) -> Optional[WorkerState]:\n         worker_state_file_path = self.state_file_path\n         if not os.path.exists(worker_state_file_path):\n", "issue": "Better handling of k8s SIGTERM when stopping workers\nWe've observed recently that worker pods can take more than 1h to get killed by kubernetes (see internal [slack thread](https://huggingface.slack.com/archives/C04L6P8KNQ5/p1697641144036899) for example).\r\nAccording to the k8s docs, at one point it sends a TERM and then after a grace period a SIGKILL on all the running processes.\r\nIf this keeps happening we need to improve the way we handle those to kill the pods faster.\r\n\r\nOne idea is to catch SIGTERM in the worker executor (main python process of workers) that would stop the worker loop (its subprocess where the actual code runs).\n", "before_files": [{"content": "# SPDX-License-Identifier: Apache-2.0\n# Copyright 2022 The HuggingFace Authors.\nimport asyncio\nimport logging\nimport os\nimport sys\nfrom collections.abc import Callable\nfrom datetime import datetime, timedelta\nfrom random import random\nfrom typing import Any, Optional, Union\n\nimport orjson\nfrom filelock import FileLock\nfrom libcommon.processing_graph import ProcessingGraph\nfrom libcommon.queue import Queue\nfrom libcommon.utils import get_datetime\nfrom mirakuru import OutputExecutor, ProcessExitedWithError\n\nfrom worker import start_worker_loop\nfrom worker.config import AppConfig\nfrom worker.job_manager import JobManager\nfrom worker.job_runner_factory import JobRunnerFactory\nfrom worker.loop import WorkerState\n\nSTART_WORKER_LOOP_PATH = start_worker_loop.__file__\n\n\nasync def every(\n    func: Callable[..., Optional[Any]],\n    *args: Any,\n    seconds: Union[float, tuple[float, float]],\n    stop_on: Optional[Any] = None,\n    **kwargs: Any,\n) -> None:\n    while True:\n        out = func(*args, **kwargs)\n        if stop_on is not None and out == stop_on:\n            break\n        delay = (\n            seconds[0] + (seconds[1] - seconds[0]) * random() if isinstance(seconds, tuple) else seconds  # nosec B311\n        )\n        await asyncio.sleep(delay)\n\n\nclass BadWorkerState(RuntimeError):\n    \"\"\"Raised when the worker state from the worker read by the executor is not valid.\"\"\"\n\n    pass\n\n\nclass WorkerExecutor:\n    def __init__(self, app_config: AppConfig, job_runner_factory: JobRunnerFactory, state_file_path: str) -> None:\n        self.app_config = app_config\n        self.job_runner_factory = job_runner_factory\n        self.state_file_path = state_file_path\n        self.processing_graph = ProcessingGraph(self.app_config.processing_graph)\n\n        max_missing_heartbeats = self.app_config.worker.max_missing_heartbeats\n        heartbeat_interval_seconds = self.app_config.worker.heartbeat_interval_seconds\n        self.max_seconds_without_heartbeat_for_zombies = heartbeat_interval_seconds * max_missing_heartbeats\n\n        self.heartbeat_interval_seconds = self.app_config.worker.heartbeat_interval_seconds\n        self.max_job_duration_seconds = self.app_config.worker.max_job_duration_seconds\n        self.kill_zombies_interval_seconds = self.app_config.worker.kill_zombies_interval_seconds\n        self.kill_long_job_interval_seconds = self.app_config.worker.kill_long_job_interval_seconds\n\n    def _create_worker_loop_executor(self) -> OutputExecutor:\n        banner = self.state_file_path\n        start_worker_loop_command = [\n            sys.executable,\n            START_WORKER_LOOP_PATH,\n            \"--print-worker-state-path\",\n        ]\n        return OutputExecutor(start_worker_loop_command, banner, timeout=10)\n\n    def start(self) -> None:\n        exceptions = []\n        worker_loop_executor = self._create_worker_loop_executor()\n        worker_loop_executor.start()  # blocking until the banner is printed\n\n        def custom_exception_handler(loop: asyncio.AbstractEventLoop, context: dict[str, Any]) -> None:\n            nonlocal exceptions\n            # first, handle with default handler\n            loop.default_exception_handler(context)\n\n            exception = context.get(\"exception\")\n            if exception:\n                exceptions.append(repr(exception))\n                loop.stop()\n\n        loop = asyncio.get_event_loop()\n        loop.set_exception_handler(custom_exception_handler)\n        logging.info(\"Starting heartbeat.\")\n        loop.create_task(every(self.heartbeat, seconds=self.heartbeat_interval_seconds))\n        loop.create_task(\n            every(\n                self.kill_zombies,\n                seconds=(\n                    self.kill_zombies_interval_seconds * 0.5,\n                    self.kill_zombies_interval_seconds * 1.5,\n                ),\n            )\n        )\n        loop.create_task(\n            every(\n                self.kill_long_job,\n                worker_loop_executor=worker_loop_executor,\n                seconds=(\n                    self.kill_long_job_interval_seconds * 0.5,\n                    self.kill_long_job_interval_seconds * 1.5,\n                ),\n            )\n        )\n        loop.run_until_complete(\n            every(self.is_worker_alive, worker_loop_executor=worker_loop_executor, seconds=1.0, stop_on=False)\n        )\n        if exceptions:\n            raise RuntimeError(f\"Some async tasks failed: {exceptions}\")\n\n    def get_state(self) -> Optional[WorkerState]:\n        worker_state_file_path = self.state_file_path\n        if not os.path.exists(worker_state_file_path):\n            return None\n        with FileLock(f\"{worker_state_file_path}.lock\"):\n            try:\n                with open(worker_state_file_path, \"rb\") as worker_state_f:\n                    worker_state = orjson.loads(worker_state_f.read())\n                    return WorkerState(\n                        current_job_info=worker_state.get(\"current_job_info\"),\n                        last_updated=datetime.fromisoformat(worker_state[\"last_updated\"]),\n                    )\n            except (orjson.JSONDecodeError, KeyError) as err:\n                raise BadWorkerState(f\"Failed to read worker state at {worker_state_file_path}\") from err\n\n    def heartbeat(self) -> None:\n        worker_state = self.get_state()\n        if worker_state and worker_state[\"current_job_info\"]:\n            Queue().heartbeat(job_id=worker_state[\"current_job_info\"][\"job_id\"])\n\n    def kill_zombies(self) -> None:\n        queue = Queue()\n        zombies = queue.get_zombies(max_seconds_without_heartbeat=self.max_seconds_without_heartbeat_for_zombies)\n        message = \"Job manager crashed while running this job (missing heartbeats).\"\n        for zombie in zombies:\n            job_runner = self.job_runner_factory.create_job_runner(zombie)\n            job_manager = JobManager(\n                job_info=zombie,\n                app_config=self.app_config,\n                job_runner=job_runner,\n                processing_graph=self.processing_graph,\n            )\n            job_manager.set_crashed(message=message)\n            logging.info(f\"Killing zombie. Job info = {zombie}\")\n\n    def kill_long_job(self, worker_loop_executor: OutputExecutor) -> None:\n        worker_state = self.get_state()\n        if worker_state and worker_state[\"current_job_info\"]:\n            long_job = worker_state[\"current_job_info\"]\n            last_updated = worker_state[\"last_updated\"]\n            coefficient = 10 if long_job[\"params\"][\"dataset\"] == \"cerebras/SlimPajama-627B\" else 1\n            if last_updated + timedelta(seconds=coefficient * self.max_job_duration_seconds) <= get_datetime():\n                _duration_seconds = int((get_datetime() - last_updated).total_seconds())\n                logging.warning(\n                    f\"Job {long_job} exceeded maximum duration of\"\n                    f\" {self.max_job_duration_seconds} seconds ({_duration_seconds} seconds).\"\n                )\n                try:\n                    worker_loop_executor.stop()  # raises an error if the worker returned exit code 1\n                finally:\n                    logging.info(f\"Killing a long job. Job info = {long_job}\")\n                    job_runner = self.job_runner_factory.create_job_runner(long_job)\n                    job_manager = JobManager(\n                        job_info=long_job,\n                        app_config=self.app_config,\n                        job_runner=job_runner,\n                        processing_graph=self.processing_graph,\n                    )\n                    message = \"Job manager was killed while running this job (job exceeded maximum duration).\"\n                    job_manager.set_exceeded_maximum_duration(message=message)\n\n    def is_worker_alive(self, worker_loop_executor: OutputExecutor) -> bool:\n        if worker_loop_executor.running():\n            return True\n        try:\n            worker_loop_executor.stop()  # raises an error if the worker returned unexpected exit code\n        except ProcessExitedWithError as err:\n            explanation = f\"exit code f{err.exit_code}\"\n            if err.exit_code == -9:\n                explanation += \" SIGKILL - surely an OOM\"\n            error_msg = f\"Worker crashed ({explanation})\"\n            state = self.get_state()\n            if state and state[\"current_job_info\"]:\n                error_msg += f\"when running job_id={state['current_job_info']['job_id']}\"\n            logging.error(error_msg)\n            raise\n        return False\n", "path": "services/worker/src/worker/executor.py"}], "after_files": [{"content": "# SPDX-License-Identifier: Apache-2.0\n# Copyright 2022 The HuggingFace Authors.\nimport asyncio\nimport logging\nimport os\nimport signal\nimport sys\nfrom collections.abc import Callable\nfrom datetime import datetime, timedelta\nfrom random import random\nfrom typing import Any, Optional, Union\n\nimport orjson\nfrom filelock import FileLock\nfrom libcommon.processing_graph import ProcessingGraph\nfrom libcommon.queue import Queue\nfrom libcommon.utils import get_datetime\nfrom mirakuru import OutputExecutor, ProcessExitedWithError\n\nfrom worker import start_worker_loop\nfrom worker.config import AppConfig\nfrom worker.job_manager import JobManager\nfrom worker.job_runner_factory import JobRunnerFactory\nfrom worker.loop import WorkerState\n\nSTART_WORKER_LOOP_PATH = start_worker_loop.__file__\n\n\nasync def every(\n    func: Callable[..., Optional[Any]],\n    *args: Any,\n    seconds: Union[float, tuple[float, float]],\n    stop_on: Optional[Any] = None,\n    **kwargs: Any,\n) -> None:\n    while True:\n        out = func(*args, **kwargs)\n        if stop_on is not None and out == stop_on:\n            break\n        delay = (\n            seconds[0] + (seconds[1] - seconds[0]) * random() if isinstance(seconds, tuple) else seconds  # nosec B311\n        )\n        await asyncio.sleep(delay)\n\n\nclass BadWorkerState(RuntimeError):\n    \"\"\"Raised when the worker state from the worker read by the executor is not valid.\"\"\"\n\n    pass\n\n\nclass WorkerExecutor:\n    def __init__(self, app_config: AppConfig, job_runner_factory: JobRunnerFactory, state_file_path: str) -> None:\n        self.app_config = app_config\n        self.job_runner_factory = job_runner_factory\n        self.state_file_path = state_file_path\n        self.processing_graph = ProcessingGraph(self.app_config.processing_graph)\n\n        max_missing_heartbeats = self.app_config.worker.max_missing_heartbeats\n        heartbeat_interval_seconds = self.app_config.worker.heartbeat_interval_seconds\n        self.max_seconds_without_heartbeat_for_zombies = heartbeat_interval_seconds * max_missing_heartbeats\n\n        self.heartbeat_interval_seconds = self.app_config.worker.heartbeat_interval_seconds\n        self.max_job_duration_seconds = self.app_config.worker.max_job_duration_seconds\n        self.kill_zombies_interval_seconds = self.app_config.worker.kill_zombies_interval_seconds\n        self.kill_long_job_interval_seconds = self.app_config.worker.kill_long_job_interval_seconds\n\n    def _create_worker_loop_executor(self) -> OutputExecutor:\n        banner = self.state_file_path\n        start_worker_loop_command = [\n            sys.executable,\n            START_WORKER_LOOP_PATH,\n            \"--print-worker-state-path\",\n        ]\n        return OutputExecutor(start_worker_loop_command, banner, timeout=10)\n\n    def start(self) -> None:\n        exceptions = []\n        worker_loop_executor = self._create_worker_loop_executor()\n        worker_loop_executor.start()  # blocking until the banner is printed\n\n        def custom_exception_handler(loop: asyncio.AbstractEventLoop, context: dict[str, Any]) -> None:\n            nonlocal exceptions\n            # first, handle with default handler\n            loop.default_exception_handler(context)\n\n            exception = context.get(\"exception\")\n            if exception:\n                exceptions.append(repr(exception))\n                loop.stop()\n\n        loop = asyncio.get_event_loop()\n        loop.add_signal_handler(signal.SIGTERM, self.stop, worker_loop_executor)\n        loop.set_exception_handler(custom_exception_handler)\n        logging.info(\"Starting heartbeat.\")\n        loop.create_task(every(self.heartbeat, seconds=self.heartbeat_interval_seconds))\n        loop.create_task(\n            every(\n                self.kill_zombies,\n                seconds=(\n                    self.kill_zombies_interval_seconds * 0.5,\n                    self.kill_zombies_interval_seconds * 1.5,\n                ),\n            )\n        )\n        loop.create_task(\n            every(\n                self.kill_long_job,\n                worker_loop_executor=worker_loop_executor,\n                seconds=(\n                    self.kill_long_job_interval_seconds * 0.5,\n                    self.kill_long_job_interval_seconds * 1.5,\n                ),\n            )\n        )\n        loop.run_until_complete(\n            every(self.is_worker_alive, worker_loop_executor=worker_loop_executor, seconds=1.0, stop_on=False)\n        )\n        if exceptions:\n            raise RuntimeError(f\"Some async tasks failed: {exceptions}\")\n\n    def stop(self, worker_loop_executor: OutputExecutor) -> None:\n        worker_loop_executor.stop()\n\n    def get_state(self) -> Optional[WorkerState]:\n        worker_state_file_path = self.state_file_path\n        if not os.path.exists(worker_state_file_path):\n            return None\n        with FileLock(f\"{worker_state_file_path}.lock\"):\n            try:\n                with open(worker_state_file_path, \"rb\") as worker_state_f:\n                    worker_state = orjson.loads(worker_state_f.read())\n                    return WorkerState(\n                        current_job_info=worker_state.get(\"current_job_info\"),\n                        last_updated=datetime.fromisoformat(worker_state[\"last_updated\"]),\n                    )\n            except (orjson.JSONDecodeError, KeyError) as err:\n                raise BadWorkerState(f\"Failed to read worker state at {worker_state_file_path}\") from err\n\n    def heartbeat(self) -> None:\n        worker_state = self.get_state()\n        if worker_state and worker_state[\"current_job_info\"]:\n            Queue().heartbeat(job_id=worker_state[\"current_job_info\"][\"job_id\"])\n\n    def kill_zombies(self) -> None:\n        queue = Queue()\n        zombies = queue.get_zombies(max_seconds_without_heartbeat=self.max_seconds_without_heartbeat_for_zombies)\n        message = \"Job manager crashed while running this job (missing heartbeats).\"\n        for zombie in zombies:\n            job_runner = self.job_runner_factory.create_job_runner(zombie)\n            job_manager = JobManager(\n                job_info=zombie,\n                app_config=self.app_config,\n                job_runner=job_runner,\n                processing_graph=self.processing_graph,\n            )\n            job_manager.set_crashed(message=message)\n            logging.info(f\"Killing zombie. Job info = {zombie}\")\n\n    def kill_long_job(self, worker_loop_executor: OutputExecutor) -> None:\n        worker_state = self.get_state()\n        if worker_state and worker_state[\"current_job_info\"]:\n            long_job = worker_state[\"current_job_info\"]\n            last_updated = worker_state[\"last_updated\"]\n            coefficient = 10 if long_job[\"params\"][\"dataset\"] == \"cerebras/SlimPajama-627B\" else 1\n            if last_updated + timedelta(seconds=coefficient * self.max_job_duration_seconds) <= get_datetime():\n                _duration_seconds = int((get_datetime() - last_updated).total_seconds())\n                logging.warning(\n                    f\"Job {long_job} exceeded maximum duration of\"\n                    f\" {self.max_job_duration_seconds} seconds ({_duration_seconds} seconds).\"\n                )\n                try:\n                    worker_loop_executor.stop()  # raises an error if the worker returned exit code 1\n                finally:\n                    logging.info(f\"Killing a long job. Job info = {long_job}\")\n                    job_runner = self.job_runner_factory.create_job_runner(long_job)\n                    job_manager = JobManager(\n                        job_info=long_job,\n                        app_config=self.app_config,\n                        job_runner=job_runner,\n                        processing_graph=self.processing_graph,\n                    )\n                    message = \"Job manager was killed while running this job (job exceeded maximum duration).\"\n                    job_manager.set_exceeded_maximum_duration(message=message)\n\n    def is_worker_alive(self, worker_loop_executor: OutputExecutor) -> bool:\n        if worker_loop_executor.running():\n            return True\n        try:\n            worker_loop_executor.stop()  # raises an error if the worker returned unexpected exit code\n        except ProcessExitedWithError as err:\n            explanation = f\"exit code f{err.exit_code}\"\n            if err.exit_code == -9:\n                explanation += \" SIGKILL - surely an OOM\"\n            error_msg = f\"Worker crashed ({explanation})\"\n            state = self.get_state()\n            if state and state[\"current_job_info\"]:\n                error_msg += f\"when running job_id={state['current_job_info']['job_id']}\"\n            logging.error(error_msg)\n            raise\n        return False\n", "path": "services/worker/src/worker/executor.py"}]}