mirror of
https://github.com/blakeblackshear/frigate.git
synced 2026-10-12 09:42:47 +03:00
Fix environment_vars not reaching ffmpeg processes (#24635)
CI / AMD64 Build (push) Waiting to run
CI / ARM Build (push) Waiting to run
CI / Jetson Jetpack 6 (push) Waiting to run
CI / AMD64 Extra Build (push) Blocked by required conditions
CI / ARM Extra Build (push) Blocked by required conditions
CI / Synaptics Build (push) Blocked by required conditions
CI / Assemble and push default build (push) Blocked by required conditions
CI / AMD64 Build (push) Waiting to run
CI / ARM Build (push) Waiting to run
CI / Jetson Jetpack 6 (push) Waiting to run
CI / AMD64 Extra Build (push) Blocked by required conditions
CI / ARM Extra Build (push) Blocked by required conditions
CI / Synaptics Build (push) Blocked by required conditions
CI / Assemble and push default build (push) Blocked by required conditions
* start the forkserver after the config applies environment_vars The forkserver started as a side effect of creating the multiprocessing Manager at the top of main(), before the config loaded, so every worker process and the ffmpeg it launched inherited the environment from before environment_vars were applied. Variables like LIBVA_DRIVER_NAME set in config.yml never reached ffmpeg. The Manager now uses the fork context, so the forkserver does not start until the first worker process. * fix test
This commit is contained in:
+3
-1
@@ -15,7 +15,9 @@ from frigate.util.config import find_config_file
|
||||
|
||||
|
||||
def main() -> None:
|
||||
manager = mp.Manager()
|
||||
# Fork this manager so the forkserver does not start until the first
|
||||
# worker process, after the config has applied environment_vars.
|
||||
manager = mp.get_context("fork").Manager()
|
||||
faulthandler.enable()
|
||||
|
||||
# Setup the logging thread
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
import multiprocessing as mp
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
import frigate.__main__ as frigate_main
|
||||
import frigate.log
|
||||
from frigate.config import FrigateConfig
|
||||
from frigate.util.process import FrigateProcess
|
||||
|
||||
ENV_KEY = "FRIGATE_TEST_PROCESS_ENV"
|
||||
|
||||
|
||||
class EnvReportingProcess(FrigateProcess):
|
||||
def __init__(self, stop_event, result_queue):
|
||||
super().__init__(stop_event, 0, name="env_reporter", daemon=True)
|
||||
self.result_queue = result_queue
|
||||
|
||||
def run(self) -> None:
|
||||
self.pre_run_setup()
|
||||
self.result_queue.put(os.environ.get(ENV_KEY))
|
||||
|
||||
|
||||
def _run_main_and_report(result_queue) -> None:
|
||||
"""Run frigate's main() with the config load and app faked out.
|
||||
|
||||
Runs in a fresh interpreter so no earlier test has started the forkserver.
|
||||
"""
|
||||
mp.set_start_method("forkserver", force=True)
|
||||
|
||||
def load(**kwargs):
|
||||
os.environ[ENV_KEY] = "applied"
|
||||
return SimpleNamespace()
|
||||
|
||||
def setup_logging(manager):
|
||||
frigate.log.log_listener = SimpleNamespace(queue=manager.Queue())
|
||||
|
||||
class FakeApp:
|
||||
def __init__(self, config, manager, stop_event):
|
||||
self.stop_event = stop_event
|
||||
|
||||
def start(self):
|
||||
child_queue = mp.Queue()
|
||||
process = EnvReportingProcess(self.stop_event, child_queue)
|
||||
process.start()
|
||||
result_queue.put(child_queue.get(timeout=30))
|
||||
process.join(timeout=10)
|
||||
|
||||
try:
|
||||
with (
|
||||
patch.object(sys, "argv", ["frigate"]),
|
||||
patch.object(FrigateConfig, "load", load),
|
||||
patch("frigate.__main__.setup_logging", setup_logging),
|
||||
patch("frigate.__main__.FrigateApp", FakeApp),
|
||||
):
|
||||
frigate_main.main()
|
||||
except Exception as err:
|
||||
result_queue.put(f"error: {err!r}")
|
||||
|
||||
|
||||
class TestWorkerEnvironment(unittest.TestCase):
|
||||
def test_env_applied_by_config_load_reaches_worker_processes(self):
|
||||
ctx = mp.get_context("spawn")
|
||||
result_queue = ctx.Queue()
|
||||
runner = ctx.Process(target=_run_main_and_report, args=(result_queue,))
|
||||
runner.start()
|
||||
|
||||
try:
|
||||
self.assertEqual(result_queue.get(timeout=60), "applied")
|
||||
finally:
|
||||
runner.join(timeout=10)
|
||||
|
||||
if runner.is_alive():
|
||||
runner.terminate()
|
||||
runner.join()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user