turbopack_node/process_pool/
mod.rs

1use std::{
2    collections::VecDeque,
3    future::Future,
4    mem::take,
5    path::{Path, PathBuf},
6    process::{ExitStatus, Stdio},
7    sync::{Arc, LazyLock},
8    time::{Duration, Instant},
9};
10
11use anyhow::{Context, Result, bail};
12use bytes::Bytes;
13use futures::join;
14use owo_colors::OwoColorize;
15use parking_lot::Mutex;
16use rustc_hash::FxHashMap;
17use tokio::{
18    io::{
19        AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader, Stderr,
20        Stdout, stderr, stdout,
21    },
22    net::{TcpListener, TcpStream},
23    process::{Child, ChildStderr, ChildStdout, Command},
24    select,
25    sync::Semaphore,
26    time::{sleep, timeout},
27};
28use turbo_rcstr::{RcStr, rcstr};
29use turbo_tasks::{FxIndexSet, ResolvedVc, Vc, duration_span};
30use turbo_tasks_fs::FileSystemPath;
31use turbopack_ecmascript::magic_identifier::unmangle_identifiers;
32
33use crate::{
34    AssetsForSourceMapping,
35    backend::{CreatePoolFuture, CreatePoolOptions, NodeBackend},
36    evaluate::{EvaluateOperation, EvaluatePool, NodeJsConnectError, Operation},
37    format::FormattingMode,
38    pool_stats::{AcquiredPermits, NodeJsPoolStats, PoolStatsSnapshot},
39    source_map::apply_source_mapping,
40};
41
42mod heap_queue;
43use heap_queue::HeapQueue;
44
45struct NodeJsPoolProcess {
46    child: Option<Child>,
47    connection: TcpStream,
48    stdout_handler: OutputStreamHandler<ChildStdout, Stdout>,
49    stderr_handler: OutputStreamHandler<ChildStderr, Stderr>,
50    /// Shared ring buffer of recent stdout lines; lives independently of
51    /// `stdout_handler` so the post-mortem on a recv failure can still read
52    /// it after the handler future has returned an error.
53    recent_stdout: RecentLines,
54    recent_stderr: RecentLines,
55    debug: bool,
56    cpu_time_invested: Duration,
57}
58
59impl Ord for NodeJsPoolProcess {
60    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
61        self.cpu_time_invested
62            .cmp(&other.cpu_time_invested)
63            .then_with(|| {
64                self.child
65                    .as_ref()
66                    .map(|c| c.id())
67                    .cmp(&other.child.as_ref().map(|c| c.id()))
68            })
69    }
70}
71
72impl PartialOrd for NodeJsPoolProcess {
73    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
74        Some(self.cmp(other))
75    }
76}
77
78impl Eq for NodeJsPoolProcess {}
79
80impl PartialEq for NodeJsPoolProcess {
81    fn eq(&self, other: &Self) -> bool {
82        self.cmp(other) == std::cmp::Ordering::Equal
83    }
84}
85
86const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
87
88/// Exit code used by the child process when it cannot connect to the port we
89/// listen on (see `js/src/child_process/index.ts`).
90const CONNECT_FAILED_EXIT_CODE: i32 = 69;
91
92#[derive(Clone, PartialEq, Eq, Hash)]
93struct OutputEntry {
94    data: Arc<[u8]>,
95    stack_trace: Option<Arc<[u8]>>,
96}
97
98type SharedOutputSet = Arc<Mutex<FxIndexSet<(OutputEntry, u32)>>>;
99
100static GLOBAL_OUTPUT_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
101static MARKER: &[u8] = b"TURBOPACK_OUTPUT_";
102static MARKER_STR: &str = "TURBOPACK_OUTPUT_";
103
104/// Maximum number of recent output lines retained per stream for diagnostic
105/// purposes — included in the error message when the subprocess crashes
106/// before completing a recv.
107const RECENT_OUTPUT_LINES: usize = 100;
108
109/// A bounded ring buffer of recent lines from a child stream. Shared with the
110/// owning [`NodeJsPoolProcess`] so that the post-mortem code on a recv failure
111/// can read what the child wrote even if the handler future has already
112/// returned an error (and dropped its local state).
113type RecentLines = Arc<Mutex<VecDeque<Vec<u8>>>>;
114
115struct OutputStreamHandler<R: AsyncRead + Unpin, W: AsyncWrite + Unpin> {
116    stream: BufReader<R>,
117    shared: SharedOutputSet,
118    assets_for_source_mapping: ResolvedVc<AssetsForSourceMapping>,
119    root: FileSystemPath,
120    project_dir: FileSystemPath,
121    final_stream: W,
122    recent_lines: RecentLines,
123}
124
125impl<R: AsyncRead + Unpin, W: AsyncWrite + Unpin> OutputStreamHandler<R, W> {
126    /// Pipes the `stream` from `final_stream`, but uses `shared` to deduplicate
127    /// lines that has beem emitted by other [OutputStreamHandler] instances
128    /// with the same `shared` before.
129    /// Returns when one operation is done.
130    pub async fn handle_operation(&mut self) -> Result<()> {
131        let Self {
132            stream,
133            shared,
134            assets_for_source_mapping,
135            root,
136            project_dir,
137            final_stream,
138            recent_lines,
139        } = self;
140
141        async fn write_final<W: AsyncWrite + Unpin>(
142            mut bytes: &[u8],
143            final_stream: &mut W,
144        ) -> Result<()> {
145            let _lock = GLOBAL_OUTPUT_LOCK.lock().await;
146            while !bytes.is_empty() {
147                let count = final_stream.write(bytes).await?;
148                if count == 0 {
149                    bail!("Failed to write to final stream as it was closed");
150                }
151                bytes = &bytes[count..];
152            }
153            Ok(())
154        }
155
156        async fn write_source_mapped_final<W: AsyncWrite + Unpin>(
157            bytes: &[u8],
158            assets_for_source_mapping: Vc<AssetsForSourceMapping>,
159            root: FileSystemPath,
160            project_dir: FileSystemPath,
161            final_stream: &mut W,
162        ) -> Result<()> {
163            if let Ok(text) = std::str::from_utf8(bytes) {
164                let text = unmangle_identifiers(text, |content| {
165                    format!("{{{content}}}").italic().to_string()
166                });
167                match apply_source_mapping(
168                    text.as_ref(),
169                    assets_for_source_mapping,
170                    root,
171                    project_dir,
172                    FormattingMode::AnsiColors,
173                )
174                .await
175                {
176                    Err(e) => {
177                        write_final(
178                            format!("Error applying source mapping: {e}\n").as_bytes(),
179                            final_stream,
180                        )
181                        .await?;
182                        write_final(text.as_bytes(), final_stream).await?;
183                    }
184                    Ok(text) => {
185                        write_final(text.as_bytes(), final_stream).await?;
186                    }
187                }
188            } else {
189                write_final(bytes, final_stream).await?;
190            }
191            Ok(())
192        }
193
194        let mut buffer = Vec::new();
195        let mut own_output = FxHashMap::default();
196        let mut nesting: u32 = 0;
197        let mut in_stack = None;
198        let mut stack_trace_buffer = Vec::new();
199        loop {
200            let start = buffer.len();
201            if stream
202                .read_until(b'\n', &mut buffer)
203                .await
204                .context("error reading from stream")?
205                == 0
206            {
207                bail!("stream closed unexpectedly")
208            }
209            // Mirror the just-read line into the recent-lines ring buffer for
210            // diagnostic capture on subprocess crash. Skip protocol markers.
211            let line_is_marker = buffer.len() - start == MARKER.len() + 2
212                && &buffer[start..buffer.len() - 2] == MARKER;
213            if !line_is_marker {
214                let mut lines = recent_lines.lock();
215                if lines.len() >= RECENT_OUTPUT_LINES {
216                    lines.pop_front();
217                }
218                lines.push_back(buffer[start..].to_vec());
219            } else {
220                // This is new line
221                buffer.pop();
222                // This is the type
223                match buffer.pop() {
224                    Some(b'B') => {
225                        stack_trace_buffer.clear();
226                        buffer.truncate(start);
227                        nesting += 1;
228                        in_stack = None;
229                        continue;
230                    }
231                    Some(b'E') => {
232                        buffer.truncate(start);
233                        if let Some(in_stack) = in_stack {
234                            if nesting != 0 {
235                                stack_trace_buffer = buffer[in_stack..].to_vec();
236                            }
237                            buffer.truncate(in_stack);
238                        }
239                        nesting = nesting.saturating_sub(1);
240                        in_stack = None;
241                        if nesting == 0 {
242                            let line = Arc::from(take(&mut buffer).into_boxed_slice());
243                            let stack_trace = if stack_trace_buffer.is_empty() {
244                                None
245                            } else {
246                                Some(Arc::from(take(&mut stack_trace_buffer).into_boxed_slice()))
247                            };
248                            let entry = OutputEntry {
249                                data: line,
250                                stack_trace,
251                            };
252                            let occurrence_number = *own_output
253                                .entry(entry.clone())
254                                .and_modify(|c| *c += 1)
255                                .or_insert(0);
256                            let new_entry = {
257                                let mut shared = shared.lock();
258                                shared.insert((entry.clone(), occurrence_number))
259                            };
260                            if !new_entry {
261                                // This line has been printed by another process, so we don't need
262                                // to print it again.
263                                continue;
264                            }
265                            write_source_mapped_final(
266                                &entry.data,
267                                **assets_for_source_mapping,
268                                root.clone(),
269                                project_dir.clone(),
270                                final_stream,
271                            )
272                            .await?;
273                        }
274                    }
275                    Some(b'S') => {
276                        buffer.truncate(start);
277                        in_stack = Some(start);
278                        continue;
279                    }
280                    Some(b'D') => {
281                        // operation done
282                        break;
283                    }
284                    _ => {}
285                }
286            }
287            if nesting != 0 {
288                // When inside of a marked output we want to aggregate until the end marker
289                continue;
290            }
291
292            write_source_mapped_final(
293                &buffer,
294                **assets_for_source_mapping,
295                root.clone(),
296                project_dir.clone(),
297                final_stream,
298            )
299            .await?;
300            buffer.clear();
301        }
302        Ok(())
303    }
304}
305
306impl NodeJsPoolProcess {
307    async fn new(
308        cwd: &Path,
309        env: &FxHashMap<RcStr, RcStr>,
310        entrypoint: &Path,
311        assets_for_source_mapping: ResolvedVc<AssetsForSourceMapping>,
312        assets_root: FileSystemPath,
313        project_dir: FileSystemPath,
314        shared_stdout: SharedOutputSet,
315        shared_stderr: SharedOutputSet,
316        debug: bool,
317    ) -> Result<Self> {
318        let guard = duration_span!("Node.js process startup", blocking = false);
319        let listener = TcpListener::bind("127.0.0.1:0")
320            .await
321            .context("binding to a port")?;
322        let port = listener.local_addr().context("getting port")?.port();
323        let mut cmd = Command::new("node");
324        cmd.current_dir(cwd);
325        if debug {
326            cmd.arg("--inspect-brk");
327        }
328        cmd.arg(entrypoint);
329        cmd.arg(port.to_string());
330        cmd.env_clear();
331        cmd.env(
332            "PATH",
333            std::env::var("PATH").expect("the PATH environment variable should always be set"),
334        );
335        #[cfg(target_family = "windows")]
336        cmd.env(
337            "SystemRoot",
338            std::env::var("SystemRoot")
339                .expect("the SystemRoot environment variable should always be set"),
340        );
341        cmd.envs(env);
342        cmd.stderr(Stdio::piped());
343        cmd.stdout(Stdio::piped());
344        cmd.kill_on_drop(true);
345
346        let mut child = cmd.spawn().context("spawning node pooled process")?;
347
348        let timeout = if debug {
349            Duration::MAX
350        } else {
351            CONNECT_TIMEOUT
352        };
353
354        async fn get_output(child: &mut Child) -> Result<(String, String)> {
355            let mut stdout = Vec::new();
356            let mut stderr = Vec::new();
357            child
358                .stdout
359                .take()
360                .unwrap()
361                .read_to_end(&mut stdout)
362                .await?;
363            child
364                .stderr
365                .take()
366                .unwrap()
367                .read_to_end(&mut stderr)
368                .await?;
369            fn clean(buffer: Vec<u8>) -> Result<String> {
370                Ok(String::from_utf8(buffer)?
371                    .lines()
372                    .filter(|line| {
373                        line.len() != MARKER_STR.len() + 1 && !line.starts_with(MARKER_STR)
374                    })
375                    .collect::<Vec<_>>()
376                    .join("\n"))
377            }
378            Ok((clean(stdout)?, clean(stderr)?))
379        }
380
381        let (connection, _) = select! {
382            connection = listener.accept() => connection.context("accepting connection")?,
383            status = child.wait() => {
384                match status {
385                    Ok(status) => {
386                        let (stdout, stderr) = get_output(&mut child).await?;
387                        if status.code() == Some(CONNECT_FAILED_EXIT_CODE) {
388                            return Err(NodeJsConnectError { stderr }.into());
389                        }
390                        bail!("node process exited before we could connect to it with {status}\nProcess output:\n{stdout}\nProcess error output:\n{stderr}");
391                    }
392                    Err(err) => {
393                        let _ = child.start_kill();
394                        let (stdout, stderr) = get_output(&mut child).await?;
395                        bail!("node process exited before we could connect to it: {err:?}\nProcess output:\n{stdout}\nProcess error output:\n{stderr}");
396                    },
397                }
398            },
399            _ = sleep(timeout) => {
400                let _ = child.start_kill();
401                let (stdout, stderr) = get_output(&mut child).await?;
402                bail!("timed out waiting for the Node.js process to connect ({timeout:?} timeout)\nProcess output:\n{stdout}\nProcess error output:\n{stderr}");
403            },
404        };
405        connection.set_nodelay(true)?;
406
407        let child_stdout = BufReader::new(child.stdout.take().unwrap());
408        let child_stderr = BufReader::new(child.stderr.take().unwrap());
409
410        let recent_stdout: RecentLines =
411            Arc::new(Mutex::new(VecDeque::with_capacity(RECENT_OUTPUT_LINES)));
412        let recent_stderr: RecentLines =
413            Arc::new(Mutex::new(VecDeque::with_capacity(RECENT_OUTPUT_LINES)));
414
415        let stdout_handler = OutputStreamHandler {
416            stream: child_stdout,
417            shared: shared_stdout,
418            assets_for_source_mapping,
419            root: assets_root.clone(),
420            project_dir: project_dir.clone(),
421            final_stream: stdout(),
422            recent_lines: recent_stdout.clone(),
423        };
424        let stderr_handler = OutputStreamHandler {
425            stream: child_stderr,
426            shared: shared_stderr,
427            assets_for_source_mapping,
428            root: assets_root.clone(),
429            project_dir: project_dir.clone(),
430            final_stream: stderr(),
431            recent_lines: recent_stderr.clone(),
432        };
433
434        let mut process = Self {
435            child: Some(child),
436            connection,
437            stdout_handler,
438            stderr_handler,
439            recent_stdout,
440            recent_stderr,
441            debug,
442            cpu_time_invested: Duration::ZERO,
443        };
444
445        drop(guard);
446
447        let guard = duration_span!("Node.js initialization", blocking = false);
448        let ready_signal = process.recv().await?;
449
450        if !ready_signal.is_empty() {
451            bail!(
452                "Node.js process didn't send the expected ready signal\nOutput:\n{}",
453                String::from_utf8_lossy(&ready_signal)
454            );
455        }
456
457        drop(guard);
458
459        Ok(process)
460    }
461
462    async fn recv(&mut self) -> Result<Bytes> {
463        let connection = &mut self.connection;
464        async fn with_timeout<T, E: Into<anyhow::Error>>(
465            debug: bool,
466            fast: bool,
467            future: impl Future<Output = Result<T, E>> + Send,
468        ) -> Result<T> {
469            if debug {
470                future.await.map_err(Into::into)
471            } else {
472                let time = if fast {
473                    Duration::from_secs(20)
474                } else {
475                    Duration::from_secs(5 * 60)
476                };
477                timeout(time, future)
478                    .await
479                    .context("timeout while receiving message from process")?
480                    .map_err(Into::into)
481            }
482        }
483        let debug = self.debug;
484        let recv_future = async move {
485            let packet_len = with_timeout(debug, false, connection.read_u32())
486                .await
487                .context("reading packet length")?
488                .try_into()
489                .context("storing packet length")?;
490            let mut packet_data = vec![0; packet_len];
491            with_timeout(debug, true, connection.read_exact(&mut packet_data))
492                .await
493                .context("reading packet data")?;
494            Ok::<_, anyhow::Error>(packet_data.into())
495        };
496        let (result, stdout, stderr) = join!(
497            recv_future,
498            self.stdout_handler.handle_operation(),
499            self.stderr_handler.handle_operation(),
500        );
501        let result = match result {
502            Err(err) => {
503                // The IPC read failed — the most common cause is the child
504                // process crashing or exiting before sending a response.
505                // Attach whatever output the child wrote (recent lines from
506                // both stream handlers + the child's exit status, if
507                // available) so the user sees something diagnostic instead
508                // of an opaque "unexpected end of file" cascade.
509                return Err(self.diagnostic_recv_error(err).await);
510            }
511            Ok(result) => result,
512        };
513        stdout.context("unable to handle stdout from the Node.js process in a structured way")?;
514        stderr.context("unable to handle stderr from the Node.js process in a structured way")?;
515        Ok(result)
516    }
517
518    /// Build a contextualized error for a failed [`Self::recv`]. Includes
519    /// the captured stdout/stderr ring buffers and, if the child has
520    /// terminated, its exit status.
521    async fn diagnostic_recv_error(&mut self, err: anyhow::Error) -> anyhow::Error {
522        let exit_status = match self.child.as_mut() {
523            Some(child) => match timeout(Duration::from_secs(2), child.wait()).await {
524                Ok(Ok(status)) => Some(status),
525                _ => None,
526            },
527            None => None,
528        };
529        let mut output = String::new();
530        let stdout_lines: Vec<Vec<u8>> = self.recent_stdout.lock().iter().cloned().collect();
531        let stderr_lines: Vec<Vec<u8>> = self.recent_stderr.lock().iter().cloned().collect();
532        if !stdout_lines.is_empty() {
533            output.push_str("\nRecent process stdout:\n");
534            for line in stdout_lines {
535                output.push_str(&String::from_utf8_lossy(&line));
536            }
537        }
538        if !stderr_lines.is_empty() {
539            output.push_str("\nRecent process stderr:\n");
540            for line in stderr_lines {
541                output.push_str(&String::from_utf8_lossy(&line));
542            }
543        }
544        let exit_note = match exit_status {
545            Some(status) => format!("Node.js process exited with {status}"),
546            None => "Node.js process is still running or its exit status is unavailable".into(),
547        };
548        // ast-grep-ignore: no-context-format
549        err.context(format!("{exit_note}{output}"))
550    }
551    async fn send(&mut self, packet_data: Bytes) -> Result<()> {
552        self.connection
553            .write_u32(
554                packet_data
555                    .len()
556                    .try_into()
557                    .context("packet length does not fit into u32")?,
558            )
559            .await
560            .context("writing packet length")?;
561        self.connection
562            .write_all(&packet_data)
563            .await
564            .context("writing packet data")?;
565        self.connection
566            .flush()
567            .await
568            .context("flushing packet data")?;
569        Ok(())
570    }
571}
572
573type IdleProcessQueues = Mutex<Vec<Arc<HeapQueue<NodeJsPoolProcess>>>>;
574
575/// All non-empty `IdleProcessQueues`s of the whole application.
576/// This is used to scale down processes globally.
577static ACTIVE_POOLS: LazyLock<IdleProcessQueues> = LazyLock::new(Default::default);
578
579/// Arguments needed to spawn a new Node.js process. Extracted so that
580/// `pre_warm` can clone them once instead of cloning each pool field
581/// individually.
582struct ProcessArgs {
583    cwd: PathBuf,
584    env: FxHashMap<RcStr, RcStr>,
585    entrypoint: PathBuf,
586    assets_for_source_mapping: ResolvedVc<AssetsForSourceMapping>,
587    assets_root: FileSystemPath,
588    project_dir: FileSystemPath,
589    shared_stdout: SharedOutputSet,
590    shared_stderr: SharedOutputSet,
591    debug: bool,
592}
593
594impl ProcessArgs {
595    async fn create_process(self) -> Result<(NodeJsPoolProcess, Duration)> {
596        let start = Instant::now();
597        let process = NodeJsPoolProcess::new(
598            &self.cwd,
599            &self.env,
600            &self.entrypoint,
601            self.assets_for_source_mapping,
602            self.assets_root,
603            self.project_dir,
604            self.shared_stdout,
605            self.shared_stderr,
606            self.debug,
607        )
608        .await?;
609        Ok((process, start.elapsed()))
610    }
611}
612
613/// A pool of Node.js workers operating on an `entrypoint` with specific `cwd` and `env`.
614///
615/// The pool will spawn processes when needed and reuses old ones. It will never spawn more then a
616/// certain number of concurrent processes. This is specified with the `concurrency` argument in the
617/// constructor.
618///
619/// The worker will *not* use the `env` of the parent process by default. All environment variables
620/// need to be provided to make the execution as pure as possible.
621#[turbo_tasks::value(
622    cell = "new",
623    serialization = "skip",
624    evict = "last",
625    eq = "manual",
626    shared
627)]
628pub struct ChildProcessPool {
629    cwd: PathBuf,
630    entrypoint: PathBuf,
631    env: FxHashMap<RcStr, RcStr>,
632    pub assets_for_source_mapping: ResolvedVc<AssetsForSourceMapping>,
633    pub assets_root: FileSystemPath,
634    pub project_dir: FileSystemPath,
635    #[turbo_tasks(unsafe_ignore, debug_ignore)]
636    idle_processes: Arc<HeapQueue<NodeJsPoolProcess>>,
637    /// Semaphore to limit the number of concurrent operations in general
638    #[turbo_tasks(unsafe_ignore, debug_ignore)]
639    concurrency_semaphore: Arc<Semaphore>,
640    /// Semaphore to limit the number of concurrently booting up processes
641    /// (excludes one-off processes)
642    #[turbo_tasks(unsafe_ignore, debug_ignore)]
643    bootup_semaphore: Arc<Semaphore>,
644    #[turbo_tasks(unsafe_ignore, debug_ignore)]
645    shared_stdout: SharedOutputSet,
646    #[turbo_tasks(unsafe_ignore, debug_ignore)]
647    shared_stderr: SharedOutputSet,
648    debug: bool,
649    #[turbo_tasks(unsafe_ignore, debug_ignore)]
650    stats: Arc<Mutex<NodeJsPoolStats>>,
651}
652
653impl ChildProcessPool {
654    /// * debug: Whether to automatically enable Node's `--inspect-brk` when spawning it. Note:
655    ///   automatically overrides concurrency to 1.
656    pub fn create(
657        cwd: PathBuf,
658        entrypoint: PathBuf,
659        env: FxHashMap<RcStr, RcStr>,
660        assets_for_source_mapping: ResolvedVc<AssetsForSourceMapping>,
661        assets_root: FileSystemPath,
662        project_dir: FileSystemPath,
663        concurrency: usize,
664        debug: bool,
665    ) -> EvaluatePool {
666        EvaluatePool::new(
667            Box::new(Self {
668                cwd,
669                entrypoint,
670                env,
671                assets_for_source_mapping,
672                assets_root: assets_root.clone(),
673                project_dir: project_dir.clone(),
674                concurrency_semaphore: Arc::new(Semaphore::new(if debug {
675                    1
676                } else {
677                    concurrency
678                })),
679                bootup_semaphore: Arc::new(Semaphore::new(1)),
680                idle_processes: Arc::new(HeapQueue::new()),
681                shared_stdout: Arc::new(Mutex::new(FxIndexSet::default())),
682                shared_stderr: Arc::new(Mutex::new(FxIndexSet::default())),
683                debug,
684                stats: Default::default(),
685            }),
686            assets_for_source_mapping,
687            assets_root,
688            project_dir,
689        )
690    }
691}
692
693#[turbo_tasks::value(shared)]
694pub(crate) struct ChildProcessesBackend;
695
696#[turbo_tasks::value_impl]
697impl NodeBackend for ChildProcessesBackend {
698    fn runtime_module_path(&self) -> RcStr {
699        rcstr!("child_process/evaluate.ts")
700    }
701
702    fn globals_module_path(&self) -> RcStr {
703        rcstr!("child_process/globals.ts")
704    }
705
706    fn create_pool(&self, options: CreatePoolOptions) -> CreatePoolFuture {
707        Box::pin(async move {
708            let CreatePoolOptions {
709                cwd,
710                entrypoint,
711                env,
712                assets_for_source_mapping,
713                assets_root,
714                project_dir,
715                concurrency,
716                debug,
717            } = options;
718
719            Ok(ChildProcessPool::create(
720                cwd,
721                entrypoint,
722                env,
723                assets_for_source_mapping,
724                assets_root,
725                project_dir,
726                concurrency,
727                debug,
728            ))
729        })
730    }
731
732    fn scale_down(&self) -> Result<()> {
733        ChildProcessPool::scale_down();
734        Ok(())
735    }
736
737    fn scale_zero(&self) -> Result<()> {
738        ChildProcessPool::scale_zero();
739        Ok(())
740    }
741}
742
743#[async_trait::async_trait]
744impl EvaluateOperation for ChildProcessPool {
745    async fn operation(&self) -> Result<Box<dyn Operation>> {
746        // Acquire a running process (handles concurrency limits, boots up the process)
747
748        let operation = {
749            let _guard = duration_span!("Node.js operation", blocking = true);
750            let (process, permits) = self.acquire_process().await?;
751            ChildProcessOperation {
752                process: Some(process),
753                permits,
754                idle_processes: self.idle_processes.clone(),
755                start: Instant::now(),
756                stats: self.stats.clone(),
757                allow_process_reuse: true,
758            }
759        };
760
761        Ok(Box::new(operation))
762    }
763
764    /// Returns a snapshot of the pool's internal statistics.
765    fn stats(&self) -> PoolStatsSnapshot {
766        self.stats.lock().snapshot()
767    }
768
769    /// Eagerly spawn a Node.js process so it's ready when the first
770    /// `operation()` is called. The process goes into the idle queue.
771    /// If a node request comes in while this is still initializing, it waits
772    /// on the bootup semaphore and will resume when the process is ready.
773    fn pre_warm(&self) {
774        let args = self.process_args();
775        let bootup_semaphore = self.bootup_semaphore.clone();
776        let idle_processes = self.idle_processes.clone();
777        let stats = self.stats.clone();
778
779        tokio::spawn(async move {
780            let Ok(bootup_permit) = bootup_semaphore.clone().acquire_owned().await else {
781                return;
782            };
783            {
784                stats.lock().add_booting_worker();
785            }
786            match args.create_process().await {
787                Ok((process, bootup_time)) => {
788                    {
789                        let mut s = stats.lock();
790                        s.add_bootup_time(bootup_time);
791                        s.finished_booting_worker();
792                    }
793                    drop(bootup_permit);
794                    idle_processes.push(process, &ACTIVE_POOLS);
795                }
796                Err(_e) => {
797                    let mut s = stats.lock();
798                    s.finished_booting_worker();
799                    s.remove_worker();
800                }
801            }
802        });
803    }
804}
805
806impl ChildProcessPool {
807    async fn acquire_process(&self) -> Result<(NodeJsPoolProcess, AcquiredPermits)> {
808        {
809            self.stats.lock().add_queued_task();
810        }
811
812        let concurrency_permit = self.concurrency_semaphore.clone().acquire_owned().await?;
813
814        let bootup = async {
815            let permit = self.bootup_semaphore.clone().acquire_owned().await;
816            let wait_time = self.stats.lock().wait_time_before_bootup();
817            tokio::time::sleep(wait_time).await;
818            permit
819        };
820
821        select! {
822            idle_process_result = self.idle_processes.pop(&ACTIVE_POOLS) => {
823                let process = idle_process_result.context("acquiring idle process permit")?;
824                Ok((process, AcquiredPermits::Idle { _concurrency_permit: concurrency_permit }))
825            },
826            bootup_permit = bootup => {
827                let bootup_permit = bootup_permit.context("acquiring bootup permit")?;
828                {
829                    self.stats.lock().add_booting_worker();
830                }
831                let (process, bootup_time) = self.create_process().await?;
832                // Update the worker count
833                {
834                    let mut stats = self.stats.lock();
835                    stats.add_bootup_time(bootup_time);
836                    stats.finished_booting_worker();
837                }
838                // Increase the allowed booting up processes
839                self.bootup_semaphore.add_permits(1);
840                Ok((process, AcquiredPermits::Fresh { _concurrency_permit: concurrency_permit, _bootup_permit: bootup_permit }))
841            }
842        }
843    }
844
845    fn process_args(&self) -> ProcessArgs {
846        ProcessArgs {
847            cwd: self.cwd.clone(),
848            env: self.env.clone(),
849            entrypoint: self.entrypoint.clone(),
850            assets_for_source_mapping: self.assets_for_source_mapping,
851            assets_root: self.assets_root.clone(),
852            project_dir: self.project_dir.clone(),
853            shared_stdout: self.shared_stdout.clone(),
854            shared_stderr: self.shared_stderr.clone(),
855            debug: self.debug,
856        }
857    }
858
859    async fn create_process(&self) -> Result<(NodeJsPoolProcess, Duration), anyhow::Error> {
860        self.process_args()
861            .create_process()
862            .await
863            .context("creating new process")
864    }
865
866    pub fn scale_down() {
867        let pools = ACTIVE_POOLS.lock().clone();
868        for pool in pools {
869            pool.reduce_to_one();
870        }
871    }
872
873    pub fn scale_zero() {
874        let pools = ACTIVE_POOLS.lock().clone();
875        for pool in pools {
876            pool.reduce_to_zero(&ACTIVE_POOLS);
877        }
878    }
879}
880
881pub struct ChildProcessOperation {
882    process: Option<NodeJsPoolProcess>,
883    // This is used for drop
884    #[allow(dead_code)]
885    permits: AcquiredPermits,
886    idle_processes: Arc<HeapQueue<NodeJsPoolProcess>>,
887    start: Instant,
888    stats: Arc<Mutex<NodeJsPoolStats>>,
889    allow_process_reuse: bool,
890}
891
892#[async_trait::async_trait]
893impl Operation for ChildProcessOperation {
894    async fn recv(&mut self) -> Result<Bytes> {
895        let bytes = self
896            .with_process(async |process| process.recv().await.context("failed to receive message"))
897            .await?;
898        Ok(bytes)
899    }
900
901    async fn send(&mut self, message: Bytes) -> Result<()> {
902        self.with_process(async |process| {
903            timeout(Duration::from_secs(30), process.send(message))
904                .await
905                .context("timeout while sending message")?
906                .context("failed to send message")?;
907            Ok(())
908        })
909        .await
910    }
911
912    async fn wait_or_kill(&mut self) -> Result<ExitStatus> {
913        let mut process = self
914            .process
915            .take()
916            .context("Node.js operation already finished")?;
917
918        if self.allow_process_reuse {
919            self.stats.lock().remove_worker();
920        }
921
922        let mut child = process
923            .child
924            .take()
925            .context("Node.js operation already finished")?;
926
927        // Ignore error since we are not sure if the process is still alive
928        let _ = child.start_kill();
929        let status = timeout(Duration::from_secs(30), child.wait())
930            .await
931            .context("timeout while waiting for process end")?
932            .context("waiting for process end")?;
933
934        Ok(status)
935    }
936
937    fn disallow_reuse(&mut self) {
938        if self.allow_process_reuse {
939            self.stats.lock().remove_worker();
940            self.allow_process_reuse = false;
941        }
942    }
943}
944
945impl ChildProcessOperation {
946    async fn with_process<'a, F: Future<Output = Result<T>> + Send + 'a, T>(
947        &'a mut self,
948        f: impl FnOnce(&'a mut NodeJsPoolProcess) -> F,
949    ) -> Result<T> {
950        let process = self
951            .process
952            .as_mut()
953            .context("Node.js operation already finished")?;
954
955        if !self.allow_process_reuse {
956            bail!("Node.js process is no longer usable");
957        }
958
959        let result = f(process).await;
960        if result.is_err() && self.allow_process_reuse {
961            self.stats.lock().remove_worker();
962            self.allow_process_reuse = false;
963        }
964        result
965    }
966}
967
968impl Drop for ChildProcessOperation {
969    fn drop(&mut self) {
970        if let Some(mut process) = self.process.take() {
971            let elapsed = self.start.elapsed();
972            {
973                let stats = &mut self.stats.lock();
974                match self.permits {
975                    AcquiredPermits::Idle { .. } => stats.add_warm_process_time(elapsed),
976                    AcquiredPermits::Fresh { .. } => stats.add_cold_process_time(elapsed),
977                }
978            }
979            if self.allow_process_reuse {
980                process.cpu_time_invested += elapsed;
981                self.idle_processes.push(process, &ACTIVE_POOLS);
982            }
983        }
984    }
985}
turbopack_node/process_pool/mod.rs

turbopack_node/process_pool/
mod.rs
