server/julia_executor.py

# Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.

"""
Local Julia Executor with Process Pool Support.

This module provides a Julia code executor with:
- Proper process cleanup on timeout (no zombie processes)
- Robust error handling and logging
- Process group management for complete cleanup
- Automatic retry on transient failures
- Optional process pool for 50-100x speedup on repeated executions

Performance Modes:
- Standard mode: Spawn new process for each execution (default for single executions)
- Pool mode: Reuse persistent Julia processes (recommended for repeated executions)
"""

from __future__ import annotations

import logging
import os
import shutil
import signal
import subprocess
import tempfile
import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

# Use julia_env hierarchy to inherit handlers from app.py's setup_logging()
logger = logging.getLogger("julia_env.executor")


@dataclass
class CodeExecResult:
    """Result of code execution."""

    stdout: str
    stderr: str
    exit_code: int


# Try to import process pool (optional dependency)
try:
    from .julia_process_pool import JuliaProcessPool

    POOL_AVAILABLE = True
except ImportError:
    POOL_AVAILABLE = False
    JuliaProcessPool = None


class JuliaExecutor:
    """
    Executor for running Julia code with robust process management.

    This class provides a safe interface to execute Julia code in isolation
    and capture the results including stdout, stderr, and exit code.

    Features:
    - Proper timeout handling without zombie processes
    - Process group cleanup for nested processes
    - Automatic retry on transient failures
    - Comprehensive logging for debugging
    - Optional process pool for 50-100x speedup on repeated executions

    Example:
        >>> executor = JuliaExecutor()
        >>> result = executor.run('println("Hello, Julia!")')
        >>> print(result.stdout)  # "Hello, Julia!\\n"
        >>> print(result.exit_code)  # 0
        >>>
        >>> # With process pool (recommended for repeated executions)
        >>> JuliaExecutor.enable_process_pool(size=4)
        >>> executor = JuliaExecutor(use_process_pool=True)
        >>> for i in range(100):
        ...     result = executor.run(f'println({i})')  # 50-100x faster!
        >>> JuliaExecutor.shutdown_pool()  # Clean up when done
    """

    # Class-level process pool (shared across all instances if enabled)
    _shared_pool: Optional["JuliaProcessPool"] = None
    _pool_lock = threading.Lock()
    _pool_size: int = 0
    _pool_timeout: int = 120

    def __init__(
        self,
        timeout: Optional[int] = None,
        max_retries: int = 0,
        use_optimization_flags: bool = True,
        use_process_pool: bool = True,
    ):
        """
        Initialize the JuliaExecutor.

        Args:
            timeout: Maximum execution time in seconds. If None, reads from
                     JULIA_EXECUTION_TIMEOUT env var (default: 120 if not set)
            max_retries: Number of retry attempts on transient failures (default: 0)
            use_optimization_flags: Enable Julia performance flags (default: True)
            use_process_pool: Use process pool if available (default: True)

        Raises:
            RuntimeError: If Julia executable is not found in PATH
        """
        # Read timeout from env var if not explicitly provided
        if timeout is None:
            timeout = int(os.getenv("JULIA_EXECUTION_TIMEOUT", "120"))
            logger.debug(
                f"Executor timeout from JULIA_EXECUTION_TIMEOUT env var: {timeout}s"
            )
        self.timeout = timeout
        self.max_retries = max_retries
        self.use_optimization_flags = use_optimization_flags
        self._use_process_pool = use_process_pool

        # Find Julia executable in PATH
        self.julia_path = shutil.which("julia")

        if not self.julia_path:
            # Try common installation paths
            common_paths = [
                os.path.expanduser("~/.juliaup/bin/julia"),
                os.path.expanduser("~/.julia/bin/julia"),
                "/usr/local/bin/julia",
                "/usr/bin/julia",
            ]

            for path in common_paths:
                if os.path.isfile(path) and os.access(path, os.X_OK):
                    self.julia_path = path
                    break

        if not self.julia_path:
            logger.warning(
                "Julia executable not found in PATH or common locations. "
                "Please install Julia: https://julialang.org/downloads/"
            )

        # Build optimized Julia command with performance flags
        self.base_cmd = [self.julia_path] if self.julia_path else ["julia"]

        if self.use_optimization_flags:
            self.base_cmd.extend(
                [
                    "--compile=min",
                    "--optimize=2",
                    "--startup-file=no",
                    "--history-file=no",
                ]
            )

        logger.debug(f"JuliaExecutor initialized with Julia at: {self.julia_path}")
        logger.debug(f"Timeout: {self.timeout}s, Max retries: {self.max_retries}")

    def _kill_process_tree(
        self, proc: subprocess.Popen, script_file: Optional[str] = None
    ) -> None:
        """
        Terminate a process and all its children.

        This prevents zombie processes by ensuring complete cleanup.

        Args:
            proc: The subprocess.Popen instance to terminate
            script_file: Optional script file path (for logging)
        """
        if proc.poll() is None:  # Process is still running
            try:
                # Try graceful termination first
                logger.warning(f"Terminating process {proc.pid} gracefully...")
                proc.terminate()

                # Wait up to 2 seconds for graceful termination
                try:
                    proc.wait(timeout=2.0)
                    logger.debug(f"Process {proc.pid} terminated gracefully")
                    return
                except subprocess.TimeoutExpired:
                    logger.warning(
                        f"Process {proc.pid} did not terminate, forcing kill..."
                    )

                # Force kill if still running
                proc.kill()
                try:
                    proc.wait(timeout=2.0)
                    logger.debug(f"Process {proc.pid} killed forcefully")
                except subprocess.TimeoutExpired:
                    pass

            except Exception as e:
                logger.error(f"Error killing process {proc.pid}: {e}")

                # Last resort: try killing via process group
                try:
                    if hasattr(os, "killpg"):
                        os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
                        logger.debug(f"Killed process group for {proc.pid}")
                except Exception as pg_error:
                    logger.error(f"Failed to kill process group: {pg_error}")

    def run(self, code: str, timeout: Optional[int] = None) -> CodeExecResult:
        """
        Execute Julia code and return the result with robust error handling.

        This method provides:
        - Automatic retry on transient failures
        - Proper timeout handling without zombie processes
        - Process group cleanup for nested processes
        - Comprehensive error logging
        - Optional process pool for 50-100x speedup

        Args:
            code: Julia code string to execute
            timeout: Override default timeout (seconds). If None, uses pool's
                     configured timeout (when using pool) or instance timeout.

        Returns:
            CodeExecResult containing stdout, stderr, and exit_code
        """
        # Use process pool if enabled and available
        # Pass timeout as-is (None means use pool's configured default)
        if self._use_process_pool and JuliaExecutor._shared_pool is not None:
            try:
                return JuliaExecutor._shared_pool.execute(code, timeout=timeout)
            except Exception as e:
                logger.warning(
                    f"Process pool execution failed: {e}, falling back to subprocess"
                )
                # Fall through to standard execution

        # For subprocess fallback, apply instance default if timeout not specified
        if timeout is None:
            timeout = self.timeout

        # Check if Julia is available
        if not self.julia_path:
            return CodeExecResult(
                stdout="",
                stderr="Julia not found in PATH. Please install Julia.",
                exit_code=127,
            )

        code_file = None

        for attempt in range(self.max_retries + 1):
            proc = None

            try:
                # Create temporary file for Julia code
                with tempfile.NamedTemporaryFile(
                    mode="w", suffix=".jl", delete=False, encoding="utf-8"
                ) as f:
                    f.write(code)
                    code_file = f.name

                script_name = Path(code_file).name
                logger.debug(
                    f"[Attempt {attempt + 1}/{self.max_retries + 1}] Executing: {script_name}"
                )

                # Start process with Popen for better control
                start_time = time.time()

                # On Unix systems, use process groups for better cleanup
                kwargs = {
                    "stdout": subprocess.PIPE,
                    "stderr": subprocess.PIPE,
                    "text": True,
                }

                # Create new process group on Unix systems
                if hasattr(os, "setpgrp"):
                    kwargs["preexec_fn"] = os.setpgrp

                proc = subprocess.Popen(self.base_cmd + [code_file], **kwargs)

                logger.debug(f"Started Julia process {proc.pid}")

                # Wait for process with timeout
                try:
                    stdout, stderr = proc.communicate(timeout=timeout)
                    exit_code = proc.returncode
                    elapsed = time.time() - start_time

                    logger.debug(
                        f"Julia execution completed in {elapsed:.2f}s (exit: {exit_code})"
                    )

                    # Clean up temp file
                    self._cleanup_temp_file(code_file)

                    return CodeExecResult(
                        stdout=stdout,
                        stderr=stderr,
                        exit_code=exit_code,
                    )

                except subprocess.TimeoutExpired:
                    logger.error(
                        f"Julia execution timed out after {timeout}s "
                        f"(attempt {attempt + 1}/{self.max_retries + 1})"
                    )

                    # CRITICAL: Kill the process AND all its children
                    self._kill_process_tree(proc, code_file)

                    # If this was our last retry, return timeout error
                    if attempt >= self.max_retries:
                        self._cleanup_temp_file(code_file)
                        return CodeExecResult(
                            stdout="",
                            stderr=f"Execution timed out after {timeout}s",
                            exit_code=124,  # Standard timeout exit code
                        )

                    # Wait before retry
                    time.sleep(1.0)
                    continue

            except FileNotFoundError:
                logger.error(f"Julia executable not found at {self.julia_path}")
                return CodeExecResult(
                    stdout="",
                    stderr=f"Julia executable not found: {self.julia_path}",
                    exit_code=127,
                )

            except Exception as e:
                logger.error(
                    f"Error executing Julia (attempt {attempt + 1}/{self.max_retries + 1}): {e}"
                )

                # Try to kill process if it exists
                if proc is not None and proc.poll() is None:
                    self._kill_process_tree(proc, code_file)

                # If this was our last retry, return error
                if attempt >= self.max_retries:
                    self._cleanup_temp_file(code_file)
                    return CodeExecResult(
                        stdout="",
                        stderr=f"Error executing Julia code: {str(e)}",
                        exit_code=1,
                    )

                # Wait before retry
                time.sleep(1.0)
                continue

            finally:
                # Always ensure temp file is cleaned up
                self._cleanup_temp_file(code_file)

        # Should never reach here
        return CodeExecResult(
            stdout="",
            stderr="Unexpected error: all retries exhausted",
            exit_code=1,
        )

    def _cleanup_temp_file(self, code_file: Optional[str]) -> None:
        """Clean up temporary file safely."""
        if code_file and Path(code_file).exists():
            try:
                Path(code_file).unlink()
            except Exception as e:
                logger.debug(f"Could not delete temp file {code_file}: {e}")

    @staticmethod
    def enable_process_pool(size: int = 4, timeout: Optional[int] = None) -> bool:
        """
        Enable the shared Julia process pool for all JuliaExecutor instances.

        This provides 50-100x speedup for repeated code executions by reusing
        persistent Julia processes instead of spawning new ones.

        Args:
            size: Number of worker processes to create (default: 4)
            timeout: Default timeout for code execution in seconds.
                     If None, reads from JULIA_EXECUTION_TIMEOUT env var (default: 120)

        Returns:
            True if pool was created successfully, False otherwise
        """
        if not POOL_AVAILABLE:
            logger.warning(
                "Process pool not available (julia_process_pool module not found). "
                "Falling back to subprocess execution."
            )
            return False

        # Read timeout from env var if not explicitly provided
        if timeout is None:
            timeout = int(os.getenv("JULIA_EXECUTION_TIMEOUT", "120"))

        with JuliaExecutor._pool_lock:
            if JuliaExecutor._shared_pool is not None:
                logger.debug("Process pool already enabled")
                return True

            try:
                logger.info(f"Enabling Julia process pool with {size} workers")
                JuliaExecutor._shared_pool = JuliaProcessPool(
                    size=size, timeout=timeout
                )
                JuliaExecutor._pool_size = size
                JuliaExecutor._pool_timeout = timeout
                logger.info("Julia process pool enabled successfully")
                return True
            except Exception as e:
                logger.error(f"Failed to enable process pool: {e}")
                return False

    @staticmethod
    def shutdown_pool() -> None:
        """
        Shutdown the shared Julia process pool.

        This should be called when you're done with all Julia executions
        to properly clean up worker processes.
        """
        with JuliaExecutor._pool_lock:
            if JuliaExecutor._shared_pool is not None:
                logger.info("Shutting down Julia process pool")
                try:
                    JuliaExecutor._shared_pool.shutdown()
                except Exception as e:
                    logger.error(f"Error shutting down pool: {e}")
                finally:
                    JuliaExecutor._shared_pool = None

    @staticmethod
    def is_pool_enabled() -> bool:
        """Check if the process pool is currently enabled."""
        with JuliaExecutor._pool_lock:
            return JuliaExecutor._shared_pool is not None

    @staticmethod
    def get_pool_metrics() -> dict:
        """Get metrics about the process pool."""
        if JuliaExecutor._shared_pool is None:
            return {
                "enabled": False,
                "pool_size": 0,
                "available_workers": 0,
            }

        return {
            "enabled": True,
            "pool_size": JuliaExecutor._pool_size,
            "timeout": JuliaExecutor._pool_timeout,
            "available_workers": JuliaExecutor._pool_size,
        }