25443604553

Committed 06 May 2026 03:05PM UTC coverage: 92.879% (-0.04%) from 92.915%

Build # 25443604553

Build Type

push

github

Committed by

web-flow

Commit Message

[pants_ng] Scaffolding for a pants_ng mode. (#23319)

In this mode the command line is parsed as an
NG invocation, and dispatched appropriately.

Of course at the moment there are no
implementations to dispatch to. That will follow.

This does expose a new option, `pants_ng` to users. 
There is a big warning not to set it, but we're not trying
to hide that we're working on a new thing, so I am
comfortable with this.

Coverage Stats

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

1294 existing lines in 76 files now uncovered.

92234 of 99306 relevant lines covered (92.88%)

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 / 25443604553

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