25403087079

Committed 05 May 2026 09:23PM UTC coverage: 92.903% (-0.04%) from 92.944%

Build # 25403087079

Build Type

Pull #23319

github

Committed by

web-flow

Commit Message

Merge 17479f77c into f46dc7805

Pull Request Pull Request #23319: [pants_ng] Scaffolding for a pants_ng mode.

Coverage Stats

25 of 76 new or added lines in 9 files covered. (32.89%)

10 existing lines in 4 files now uncovered.

91968 of 98994 relevant lines covered (92.9%)

4.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

27.06

/src/python/pants/bin/daemon_pants_runner.py

# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).

import logging
import os
import sys
import time
from contextlib import contextmanager
from threading import Lock

from pants.base.exiter import PANTS_FAILED_EXIT_CODE, ExitCode
from pants.bin.local_pants_runner import LocalPantsRunner
from pants.engine.env_vars import CompleteEnvironmentVars
from pants.engine.internals.native_engine import PyNgInvocation, PySessionCancellationLatch
from pants.init.logging import stdio_destination
from pants.option.options_bootstrapper import OptionsBootstrapper
from pants.pantsd.pants_daemon_core import PantsDaemonCore

logger = logging.getLogger(__name__)


class ExclusiveRequestTimeout(Exception):
    """Represents a timeout while waiting for another request to complete."""


class DaemonPantsRunner:
    """A RawFdRunner (callable) that will be called for each client request to Pantsd."""

    def __init__(self, core: PantsDaemonCore) -> None:
        super().__init__()
        self._core = core
        self._run_lock = Lock()

    @staticmethod
    def _send_stderr(stderr_fileno: int, msg: str) -> None:
        """Used to send stderr on a raw filehandle _before_ stdio replacement.

        TODO: This method will be removed as part of #7654.
        """
        with os.fdopen(stderr_fileno, mode="w", closefd=False) as stderr:
            print(msg, file=stderr, flush=True)

    @contextmanager
    def _one_run_at_a_time(
        self, stderr_fileno: int, cancellation_latch: PySessionCancellationLatch, timeout: float
    ):
        """Acquires exclusive access within the daemon.

        Periodically prints a message on the given stderr_fileno while exclusive access cannot be
        acquired.

        TODO: This method will be removed as part of #7654, so it currently polls the lock and
        cancellation latch rather than waiting for both of them asynchronously, which would be a bit
        cleaner.
        """

        render_timeout = 5
        should_poll_forever = timeout <= 0
        start = time.time()
        render_deadline = start + render_timeout
        deadline = None if should_poll_forever else start + timeout

        def should_keep_polling(now):
            return not cancellation_latch.is_cancelled() and (not deadline or deadline > now)

        acquired = self._run_lock.acquire(blocking=False)
        if not acquired:
            # If we don't acquire immediately, send an explanation.
            length = "forever" if should_poll_forever else f"up to {timeout} seconds"
            self._send_stderr(
                stderr_fileno,
                f"Another pants invocation is running. Will wait {length} for it to finish before giving up.\n"
                "If you don't want to wait for the first run to finish, please press Ctrl-C and run "
                "this command with PANTS_CONCURRENT=True in the environment.\n",
            )
        while True:
            now = time.time()
            if acquired:
                try:
                    yield
                    break
                finally:
                    self._run_lock.release()
            elif should_keep_polling(now):
                if now > render_deadline:
                    self._send_stderr(
                        stderr_fileno,
                        f"Waiting for invocation to finish (waited for {int(now - start)}s so far)...\n",
                    )
                    render_deadline = now + render_timeout
                acquired = self._run_lock.acquire(blocking=True, timeout=0.1)
            else:
                raise ExclusiveRequestTimeout(
                    "Timed out while waiting for another pants invocation to finish."
                )

    def single_daemonized_run(
        self,
        args: tuple[str, ...],
        env: dict[str, str],
        working_dir: str,
        cancellation_latch: PySessionCancellationLatch,
    ) -> ExitCode:
        """Run a single daemonized run of Pants.

        All aspects of the `sys` global should already have been replaced in `__call__`, so this
        method should not need any special handling for the fact that it's running in a proxied
        environment.
        """
        # Create a transient OptionsBootstrapper with no args, to read the value of pants_ng from
        # config and env. We can't parse args until we know if we're ng or og.
        options_bootstrapper = OptionsBootstrapper.create(args=[], env=env, allow_pantsrc=True)
        pants_ng = options_bootstrapper.bootstrap_options.for_global_scope().pants_ng

        try:
            logger.debug("Connected to pantsd")
            logger.debug(f"work dir: {working_dir}")

            # Capture the client's start time, which we propagate here in order to get an accurate
            # view of total time.
            env_start_time = env.get("PANTSD_RUNTRACKER_CLIENT_START_TIME", None)
            if not env_start_time:
                # NB: We warn rather than erroring here because it eases use of non-Pants nailgun
                # clients for testing.
                logger.warning(
                    "No start time was reported by the client! Metrics may be inaccurate."
                )
            start_time = float(env_start_time) if env_start_time else time.time()

            if pants_ng:
                logger.info("DaemonPantsRunner running as pants_ng")
                ng_invocation = PyNgInvocation.from_args(args[1:])
                # Allow the existing logic to read flags in global position (note, these can be
                # prefixed with a scope so they are not necessarily just global options), so
                # that the engine can configure itself.
                global_flags = ng_invocation.global_flag_strings()
                options_bootstrapper = OptionsBootstrapper.create(
                    args=[args[0], *global_flags], env=env, allow_pantsrc=True
                )
            else:
                ng_invocation = None
                options_bootstrapper = OptionsBootstrapper.create(
                    args=args, env=env, allow_pantsrc=True
                )

            # Run using the pre-warmed Session.
            complete_env = CompleteEnvironmentVars(env)
            scheduler, options_initializer = self._core.prepare(options_bootstrapper, complete_env)
            runner = LocalPantsRunner.create(
                complete_env,
                working_dir,
                options_bootstrapper,
                scheduler=scheduler,
                options_initializer=options_initializer,
                cancellation_latch=cancellation_latch,
                ng_invocation=ng_invocation,
            )
            return runner.run(start_time)
        except Exception as e:
            logger.exception(e)
            return PANTS_FAILED_EXIT_CODE
        except KeyboardInterrupt:
            print("Interrupted by user.\n", file=sys.stderr)
            return PANTS_FAILED_EXIT_CODE

    def __call__(
        self,
        command: str,
        args: tuple[str, ...],
        env: dict[str, str],
        working_dir: str,
        cancellation_latch: PySessionCancellationLatch,
        stdin_fileno: int,
        stdout_fileno: int,
        stderr_fileno: int,
    ) -> ExitCode:
        request_timeout = float(env.get("PANTSD_REQUEST_TIMEOUT_LIMIT", -1))
        # NB: Order matters: we acquire a lock before mutating either `sys.std*`, `os.environ`, etc.
        with self._one_run_at_a_time(
            stderr_fileno,
            cancellation_latch=cancellation_latch,
            timeout=request_timeout,
        ):
            # NB: `single_daemonized_run` implements exception handling, so only the most primitive
            # errors will escape this function, where they will be logged by the server.
            logger.info(f"handling request: `{' '.join(args)}`")
            try:
                with stdio_destination(
                    stdin_fileno=stdin_fileno,
                    stdout_fileno=stdout_fileno,
                    stderr_fileno=stderr_fileno,
                ):
                    return self.single_daemonized_run(
                        ((command,) + args), env, working_dir, cancellation_latch
                    )
            finally:
                logger.info(f"request completed: `{' '.join(args)}`")

1	# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3
4	import logging	12✔
5	import os	12✔
6	import sys	12✔
7	import time	12✔
8	from contextlib import contextmanager	12✔
9	from threading import Lock	12✔
10
11	from pants.base.exiter import PANTS_FAILED_EXIT_CODE, ExitCode	12✔
12	from pants.bin.local_pants_runner import LocalPantsRunner	12✔
13	from pants.engine.env_vars import CompleteEnvironmentVars	12✔
14	from pants.engine.internals.native_engine import PyNgInvocation, PySessionCancellationLatch	12✔
15	from pants.init.logging import stdio_destination	12✔
16	from pants.option.options_bootstrapper import OptionsBootstrapper	12✔
17	from pants.pantsd.pants_daemon_core import PantsDaemonCore	12✔
18
19	logger = logging.getLogger(__name__)	12✔
20
21
22	class ExclusiveRequestTimeout(Exception):	12✔
23	"""Represents a timeout while waiting for another request to complete."""
24
25
26	class DaemonPantsRunner:	12✔
27	"""A RawFdRunner (callable) that will be called for each client request to Pantsd."""
28
29	def __init__(self, core: PantsDaemonCore) -> None:	12✔
30	super().__init__()	×
31	self._core = core	×
32	self._run_lock = Lock()	×
33
34	@staticmethod	12✔
35	def _send_stderr(stderr_fileno: int, msg: str) -> None:	12✔
36	"""Used to send stderr on a raw filehandle _before_ stdio replacement.
37
38	TODO: This method will be removed as part of #7654.
39	"""
40	with os.fdopen(stderr_fileno, mode="w", closefd=False) as stderr:	×
41	print(msg, file=stderr, flush=True)	×
42
43	@contextmanager	12✔
44	def _one_run_at_a_time(	12✔
45	self, stderr_fileno: int, cancellation_latch: PySessionCancellationLatch, timeout: float
46	):
47	"""Acquires exclusive access within the daemon.
48
49	Periodically prints a message on the given stderr_fileno while exclusive access cannot be
50	acquired.
51
52	TODO: This method will be removed as part of #7654, so it currently polls the lock and
53	cancellation latch rather than waiting for both of them asynchronously, which would be a bit
54	cleaner.
55	"""
56
57	render_timeout = 5	×
58	should_poll_forever = timeout <= 0	×
59	start = time.time()	×
60	render_deadline = start + render_timeout	×
61	deadline = None if should_poll_forever else start + timeout	×
62
63	def should_keep_polling(now):	×
64	return not cancellation_latch.is_cancelled() and (not deadline or deadline > now)	×
65
66	acquired = self._run_lock.acquire(blocking=False)	×
67	if not acquired:	×
68	# If we don't acquire immediately, send an explanation.
69	length = "forever" if should_poll_forever else f"up to {timeout} seconds"	×
70	self._send_stderr(	×
71	stderr_fileno,
72	f"Another pants invocation is running. Will wait {length} for it to finish before giving up.\n"
73	"If you don't want to wait for the first run to finish, please press Ctrl-C and run "
74	"this command with PANTS_CONCURRENT=True in the environment.\n",
75	)
76	while True:	×
77	now = time.time()	×
78	if acquired:	×
79	try:	×
80	yield	×
81	break	×
82	finally:
83	self._run_lock.release()	×
84	elif should_keep_polling(now):	×
85	if now > render_deadline:	×
86	self._send_stderr(	×
87	stderr_fileno,
88	f"Waiting for invocation to finish (waited for {int(now - start)}s so far)...\n",
89	)
90	render_deadline = now + render_timeout	×
91	acquired = self._run_lock.acquire(blocking=True, timeout=0.1)	×
92	else:
93	raise ExclusiveRequestTimeout(	×
94	"Timed out while waiting for another pants invocation to finish."
95	)
96
97	def single_daemonized_run(	12✔
98	self,
99	args: tuple[str, ...],
100	env: dict[str, str],
101	working_dir: str,
102	cancellation_latch: PySessionCancellationLatch,
103	) -> ExitCode:
104	"""Run a single daemonized run of Pants.
105
106	All aspects of the `sys` global should already have been replaced in `__call__`, so this
107	method should not need any special handling for the fact that it's running in a proxied
108	environment.
109	"""
110	# Create a transient OptionsBootstrapper with no args, to read the value of pants_ng from
111	# config and env. We can't parse args until we know if we're ng or og.
NEW 112	options_bootstrapper = OptionsBootstrapper.create(args=[], env=env, allow_pantsrc=True)	×
NEW 113	pants_ng = options_bootstrapper.bootstrap_options.for_global_scope().pants_ng	×
114
115	try:	×
116	logger.debug("Connected to pantsd")	×
117	logger.debug(f"work dir: {working_dir}")	×
118
119	# Capture the client's start time, which we propagate here in order to get an accurate
120	# view of total time.
121	env_start_time = env.get("PANTSD_RUNTRACKER_CLIENT_START_TIME", None)	×
122	if not env_start_time:	×
123	# NB: We warn rather than erroring here because it eases use of non-Pants nailgun
124	# clients for testing.
125	logger.warning(	×
126	"No start time was reported by the client! Metrics may be inaccurate."
127	)
128	start_time = float(env_start_time) if env_start_time else time.time()	×
129
NEW 130	if pants_ng:	×
NEW 131	logger.info("DaemonPantsRunner running as pants_ng")	×
NEW 132	ng_invocation = PyNgInvocation.from_args(args[1:])	×
133	# Allow the existing logic to read flags in global position (note, these can be
134	# prefixed with a scope so they are not necessarily just global options), so
135	# that the engine can configure itself.
NEW 136	global_flags = ng_invocation.global_flag_strings()	×
NEW 137	options_bootstrapper = OptionsBootstrapper.create(	×
138	args=[args[0], *global_flags], env=env, allow_pantsrc=True
139	)
140	else:
NEW 141	ng_invocation = None	×
NEW 142	options_bootstrapper = OptionsBootstrapper.create(	×
143	args=args, env=env, allow_pantsrc=True
144	)
145
146	# Run using the pre-warmed Session.
147	complete_env = CompleteEnvironmentVars(env)	×
148	scheduler, options_initializer = self._core.prepare(options_bootstrapper, complete_env)	×
149	runner = LocalPantsRunner.create(	×
150	complete_env,
151	working_dir,
152	options_bootstrapper,
153	scheduler=scheduler,
154	options_initializer=options_initializer,
155	cancellation_latch=cancellation_latch,
156	ng_invocation=ng_invocation,
157	)
158	return runner.run(start_time)	×
159	except Exception as e:	×
160	logger.exception(e)	×
161	return PANTS_FAILED_EXIT_CODE	×
162	except KeyboardInterrupt:	×
163	print("Interrupted by user.\n", file=sys.stderr)	×
164	return PANTS_FAILED_EXIT_CODE	×
165
166	def __call__(	12✔
167	self,
168	command: str,
169	args: tuple[str, ...],
170	env: dict[str, str],
171	working_dir: str,
172	cancellation_latch: PySessionCancellationLatch,
173	stdin_fileno: int,
174	stdout_fileno: int,
175	stderr_fileno: int,
176	) -> ExitCode:
177	request_timeout = float(env.get("PANTSD_REQUEST_TIMEOUT_LIMIT", -1))	×
178	# NB: Order matters: we acquire a lock before mutating either `sys.std*`, `os.environ`, etc.
179	with self._one_run_at_a_time(	×
180	stderr_fileno,
181	cancellation_latch=cancellation_latch,
182	timeout=request_timeout,
183	):
184	# NB: `single_daemonized_run` implements exception handling, so only the most primitive
185	# errors will escape this function, where they will be logged by the server.
186	logger.info(f"handling request: `{' '.join(args)}`")	×
187	try:	×
188	with stdio_destination(	×
189	stdin_fileno=stdin_fileno,
190	stdout_fileno=stdout_fileno,
191	stderr_fileno=stderr_fileno,
192	):
193	return self.single_daemonized_run(	×
194	((command,) + args), env, working_dir, cancellation_latch
195	)
196	finally:
197	logger.info(f"request completed: `{' '.join(args)}`")	×

pantsbuild / pants / 25403087079

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous