bext_php/

pool.rs

//! PHP Worker Pool — dedicated OS threads for PHP script execution.
//!
//! Supports two modes:
//! 1. Classic mode: one `php_execute_script()` per HTTP request
//! 2. Worker mode:  boot a worker script once, dispatch requests to its
//!    `bext_handle_request()` loop (eliminates framework bootstrap)
//!
//! Architecture mirrors `bext-core::jsc_ssr::pool::JscRenderPool`.

use super::context::RequestCtx;
use super::ffi;
use std::cell::RefCell;
use std::ffi::CString;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

// ─── Thread-local worker state (for worker mode) ─────────────────────────

/// Opaque pointer to a heap-allocated `RequestCtx`, safe to send through channels.
///
/// # Safety
///
/// Ownership is transferred linearly: the producer creates a `Box<RequestCtx>`,
/// converts it to a raw pointer, wraps it in `SendPtr`, and sends it through a
/// bounded channel.  The consumer receives exclusive ownership and must either
/// convert it back to `Box<RequestCtx>` via `Box::from_raw` or ensure the C
/// side doesn't access it concurrently.  The pointer is never aliased — only
/// one thread holds it at any time.
pub(crate) struct SendPtr(pub(crate) *mut u8);
unsafe impl Send for SendPtr {}

/// Per-thread state used by the FFI callbacks in worker mode.
pub(crate) struct WorkerThreadState {
    /// Receives request context pointers from the pool dispatcher.
    pub(crate) request_rx: flume::Receiver<SendPtr>,
    /// Signals that the current request is complete.
    pub(crate) done_tx: flume::Sender<()>,
}

thread_local! {
    pub static WORKER_THREAD_STATE: RefCell<Option<WorkerThreadState>> = const { RefCell::new(None) };
}

// ─── Request / Response types ────────────────────────────────────────────

/// A PHP execution request dispatched to a worker.
pub enum PhpRequest {
    /// Execute a PHP script and return the output.
    Execute(Box<PhpExecuteRequest>),
    Shutdown,
}

/// Inner data for a PHP execute request (boxed to reduce enum size).
pub struct PhpExecuteRequest {
    pub script_path: String,
    pub method: String,
    pub uri: String,
    pub query_string: String,
    pub content_type: Option<String>,
    pub body: Vec<u8>,
    pub cookies: Option<String>,
    pub headers: Vec<(String, String)>,
    pub remote_addr: Option<String>,
    pub server_name: Option<String>,
    pub server_port: u16,
    pub https: bool,
    pub reply: flume::Sender<PhpResponse>,
}

/// Response from a PHP worker.
#[derive(Debug, Clone)]
pub enum PhpResponse {
    Ok {
        status: u16,
        body: Vec<u8>,
        headers: Vec<(String, String)>,
        exec_time_us: u64,
    },
    Error(String),
}

/// Pool-level statistics.
#[derive(Debug, Clone, Default)]
pub struct PhpPoolStats {
    pub workers: u32,
    pub active: u32,
    pub total_requests: u64,
    pub total_errors: u64,
    pub avg_exec_time_us: u64,
}

// ─── Execution mode ──────────────────────────────────────────────────────

/// How PHP scripts are executed.
#[derive(Debug, Clone)]
pub enum PhpMode {
    /// One `php_execute_script()` per request. Simple, compatible with everything.
    Classic,
    /// Boot a worker script once; dispatch requests to its `bext_handle_request()` loop.
    /// Eliminates per-request framework bootstrap (~3ms for Laravel).
    Worker {
        /// Path to the worker PHP script.
        script: String,
    },
}

// ─── Pool ────────────────────────────────────────────────────────────────

pub struct PhpPool {
    sender: flume::Sender<PhpRequest>,
    workers: Vec<std::thread::JoinHandle<()>>,
    active: Arc<AtomicU32>,
    total_requests: Arc<AtomicU64>,
    total_errors: Arc<AtomicU64>,
    total_exec_time_us: Arc<AtomicU64>,
    worker_count: u32,
    mode: PhpMode,
}

impl PhpPool {
    /// Create a classic-mode pool.
    pub fn new(worker_count: usize) -> Result<Self, String> {
        Self::create(worker_count, 0, PhpMode::Classic, None)
    }

    /// Create a classic-mode pool with per-worker request lifecycle limit.
    pub fn with_max_requests(worker_count: usize, max_requests: u64) -> Result<Self, String> {
        Self::create(worker_count, max_requests, PhpMode::Classic, None)
    }

    /// Create a worker-mode pool. The worker script boots the framework
    /// once and calls `bext_handle_request($callback)` in a loop.
    pub fn worker(
        worker_count: usize,
        worker_script: String,
        max_requests: u64,
    ) -> Result<Self, String> {
        Self::create(
            worker_count,
            max_requests,
            PhpMode::Worker {
                script: worker_script.clone(),
            },
            Some(worker_script),
        )
    }

    fn create(
        worker_count: usize,
        max_requests: u64,
        mode: PhpMode,
        worker_script: Option<String>,
    ) -> Result<Self, String> {
        // Queue depth = 128 × worker count.  Requests beyond this will block
        // the caller (backpressure), matching PHP-FPM's behavior under load.
        let (sender, receiver) = flume::bounded::<PhpRequest>(worker_count.max(1) * 128);
        let active = Arc::new(AtomicU32::new(0));
        let total_requests = Arc::new(AtomicU64::new(0));
        let total_errors = Arc::new(AtomicU64::new(0));
        let total_exec_time_us = Arc::new(AtomicU64::new(0));

        let mut workers = Vec::with_capacity(worker_count);

        for i in 0..worker_count {
            let rx = receiver.clone();
            let active = Arc::clone(&active);
            let total_requests = Arc::clone(&total_requests);
            let total_errors = Arc::clone(&total_errors);
            let total_exec_time_us = Arc::clone(&total_exec_time_us);
            let ws = worker_script.clone();

            let handle = std::thread::Builder::new()
                .name(format!("bext-php-{}", i))
                // PHP 8.4's stack checker (zend.max_allowed_stack_size) auto-detects
                // from getrlimit which reports the MAIN thread's limit, not ours.
                // Use 16MB to ensure plenty of room for PHP's compiler + JIT.
                .stack_size(16 * 1024 * 1024)
                .spawn(move || {
                    // Wrap in catch_unwind for crash recovery
                    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                        if let Some(ref script) = ws {
                            worker_mode_loop(
                                i,
                                rx,
                                active,
                                total_requests,
                                total_errors,
                                total_exec_time_us,
                                script,
                                max_requests,
                            );
                        } else {
                            classic_mode_loop(
                                i,
                                rx,
                                active,
                                total_requests,
                                total_errors,
                                total_exec_time_us,
                                max_requests,
                            );
                        }
                    }));
                    if let Err(e) = result {
                        let msg = if let Some(s) = e.downcast_ref::<String>() {
                            s.clone()
                        } else if let Some(s) = e.downcast_ref::<&str>() {
                            s.to_string()
                        } else {
                            "unknown panic".to_string()
                        };
                        tracing::error!(worker = i, error = %msg, "PHP worker panicked");
                    }
                })
                .map_err(|e| format!("Failed to spawn PHP worker {}: {}", i, e))?;

            workers.push(handle);
        }

        Ok(Self {
            sender,
            workers,
            active,
            total_requests,
            total_errors,
            total_exec_time_us,
            worker_count: worker_count as u32,
            mode,
        })
    }

    /// Execute a PHP request. Blocks until a worker completes.
    #[allow(clippy::too_many_arguments)]
    pub fn execute(
        &self,
        script_path: String,
        method: String,
        uri: String,
        query_string: String,
        content_type: Option<String>,
        body: Vec<u8>,
        cookies: Option<String>,
        headers: Vec<(String, String)>,
        remote_addr: Option<String>,
        server_name: Option<String>,
        server_port: u16,
        https: bool,
    ) -> Result<PhpResponse, String> {
        let (tx, rx) = flume::bounded(1);
        self.sender
            .send_timeout(
                PhpRequest::Execute(Box::new(PhpExecuteRequest {
                    script_path,
                    method,
                    uri,
                    query_string,
                    content_type,
                    body,
                    cookies,
                    headers,
                    remote_addr,
                    server_name,
                    server_port,
                    https,
                    reply: tx,
                })),
                Duration::from_secs(30),
            )
            .map_err(|_| "PHP pool queue timeout (30s)".to_string())?;
        rx.recv_timeout(Duration::from_secs(60))
            .map_err(|_| "PHP worker timeout (60s)".to_string())
    }

    pub fn healthy_workers(&self) -> u32 {
        self.workers.iter().filter(|h| !h.is_finished()).count() as u32
    }

    pub fn stats(&self) -> PhpPoolStats {
        let total = self.total_requests.load(Ordering::Relaxed);
        PhpPoolStats {
            workers: self.worker_count,
            active: self.active.load(Ordering::Relaxed),
            total_requests: total,
            total_errors: self.total_errors.load(Ordering::Relaxed),
            avg_exec_time_us: if total > 0 {
                self.total_exec_time_us.load(Ordering::Relaxed) / total
            } else {
                0
            },
        }
    }

    pub fn mode(&self) -> &PhpMode {
        &self.mode
    }

    pub fn shutdown(self) {
        for _ in &self.workers {
            let _ = self.sender.send(PhpRequest::Shutdown);
        }
        for handle in self.workers {
            let _ = handle.join();
        }
    }
}

const WORKER_RECV_TIMEOUT: Duration = Duration::from_secs(30);

// ─── Classic mode worker loop ────────────────────────────────────────────

fn classic_mode_loop(
    worker_id: usize,
    rx: flume::Receiver<PhpRequest>,
    active: Arc<AtomicU32>,
    total_requests: Arc<AtomicU64>,
    total_errors: Arc<AtomicU64>,
    total_exec_time_us: Arc<AtomicU64>,
    max_requests: u64,
) {
    tracing::info!(worker = worker_id, mode = "classic", "PHP worker started");
    let mut local_count: u64 = 0;

    loop {
        if max_requests > 0 && local_count >= max_requests {
            tracing::info!(
                worker = worker_id,
                requests = local_count,
                "PHP worker rotating"
            );
            break;
        }

        let request = match rx.recv_timeout(WORKER_RECV_TIMEOUT) {
            Ok(req) => req,
            Err(flume::RecvTimeoutError::Timeout) => continue,
            Err(flume::RecvTimeoutError::Disconnected) => break,
        };

        match request {
            PhpRequest::Shutdown => break,
            PhpRequest::Execute(req) => {
                active.fetch_add(1, Ordering::Relaxed);
                total_requests.fetch_add(1, Ordering::Relaxed);
                local_count += 1;

                let response = execute_classic(
                    &req.script_path,
                    &req.method,
                    &req.uri,
                    &req.query_string,
                    req.content_type.as_deref(),
                    req.body,
                    req.cookies.as_deref(),
                    req.headers,
                    req.remote_addr.as_deref(),
                    req.server_name.as_deref(),
                    req.server_port,
                    req.https,
                );

                match &response {
                    PhpResponse::Ok { exec_time_us, .. } => {
                        total_exec_time_us.fetch_add(*exec_time_us, Ordering::Relaxed);
                    }
                    PhpResponse::Error(_) => {
                        total_errors.fetch_add(1, Ordering::Relaxed);
                    }
                }

                let _ = req.reply.send(response);
                active.fetch_sub(1, Ordering::Relaxed);
            }
        }
    }

    tracing::info!(worker = worker_id, "PHP worker stopped");
}

// ─── Worker mode loop ────────────────────────────────────────────────────

#[allow(clippy::too_many_arguments)]
fn worker_mode_loop(
    worker_id: usize,
    rx: flume::Receiver<PhpRequest>,
    active: Arc<AtomicU32>,
    total_requests: Arc<AtomicU64>,
    total_errors: Arc<AtomicU64>,
    total_exec_time_us: Arc<AtomicU64>,
    worker_script: &str,
    _max_requests: u64,
) {
    tracing::info!(worker = worker_id, mode = "worker", script = %worker_script, "PHP worker started");

    // Create channels for the C↔Rust worker protocol:
    // - request_tx/rx: pool sends request ctx pointers to the C bext_handle_request()
    // - done_tx/rx: C signals request completion back to pool
    let (request_tx, request_rx) = flume::bounded::<SendPtr>(1);
    let (done_tx, done_rx) = flume::bounded::<()>(1);

    // Install thread-local state for FFI callbacks
    WORKER_THREAD_STATE.with(|state| {
        *state.borrow_mut() = Some(WorkerThreadState {
            request_rx,
            done_tx,
        });
    });

    // Boot the worker script in a background thread.
    // The script runs `while (bext_handle_request($cb)) { ... }` which blocks
    // in bext_sapi_worker_wait_request() until we send a request.
    let c_script = match CString::new(worker_script) {
        Ok(s) => s,
        Err(e) => {
            tracing::error!(worker = worker_id, error = %e, "Invalid worker script path");
            return;
        }
    };

    // Create a minimal initial context for the boot phase
    let mut boot_ctx = RequestCtx::new(Vec::new(), None, Vec::new(), None, None, 80, false);

    // Launch PHP execution on THIS thread (it will block in the bext_handle_request loop)
    // We use a separate thread to dispatch requests to it.
    let dispatcher_rx = rx;
    let dispatcher_active = active;
    let dispatcher_total_requests = total_requests;
    let dispatcher_total_errors = total_errors;
    let dispatcher_total_exec_time_us = total_exec_time_us;

    // Spawn a dispatcher thread that receives PhpRequests and feeds them to the worker
    let request_tx_clone = request_tx.clone();
    let dispatcher = std::thread::Builder::new()
        .name(format!("bext-php-{}-dispatch", worker_id))
        .spawn(move || {
            loop {
                // Drain any stale completion signal from a previous timed-out request.
                // After a timeout, the PHP worker may still complete and send on done_tx,
                // leaving a stale signal that would cause the next iteration to reclaim
                // the wrong ctx_ptr (use-after-free).
                while done_rx.try_recv().is_ok() {}

                let request = match dispatcher_rx.recv_timeout(WORKER_RECV_TIMEOUT) {
                    Ok(req) => req,
                    Err(flume::RecvTimeoutError::Timeout) => continue,
                    Err(flume::RecvTimeoutError::Disconnected) => break,
                };

                match request {
                    PhpRequest::Shutdown => {
                        // Signal the worker to exit by dropping the request channel
                        drop(request_tx_clone);
                        break;
                    }
                    PhpRequest::Execute(req) => {
                        dispatcher_active.fetch_add(1, Ordering::Relaxed);
                        dispatcher_total_requests.fetch_add(1, Ordering::Relaxed);
                        let start = Instant::now();

                        // Build a request context and send its pointer to the worker
                        let mut req_ctx = Box::new(RequestCtx::new(
                            req.body,
                            req.cookies.as_deref(),
                            req.headers,
                            req.remote_addr.as_deref(),
                            req.server_name.as_deref(),
                            req.server_port,
                            req.https,
                        ));
                        req_ctx.set_request_info(
                            &req.method,
                            &req.uri,
                            &req.query_string,
                            req.content_type.as_deref(),
                        );

                        let ctx_ptr = Box::into_raw(req_ctx) as *mut u8;

                        // Send to the PHP worker (blocks until bext_handle_request picks it up)
                        if request_tx_clone.send(SendPtr(ctx_ptr)).is_err() {
                            // Worker died — convert back to Box to drop
                            let _ = unsafe { Box::from_raw(ctx_ptr as *mut RequestCtx) };
                            let _ = req.reply.send(PhpResponse::Error("Worker died".into()));
                            dispatcher_active.fetch_sub(1, Ordering::Relaxed);
                            dispatcher_total_errors.fetch_add(1, Ordering::Relaxed);
                            break;
                        }

                        // Wait for the worker to finish processing
                        match done_rx.recv_timeout(Duration::from_secs(60)) {
                            Ok(()) => {
                                // Reclaim the context and extract the response
                                let req_ctx = unsafe { Box::from_raw(ctx_ptr as *mut RequestCtx) };
                                let exec_time_us = start.elapsed().as_micros() as u64;
                                dispatcher_total_exec_time_us
                                    .fetch_add(exec_time_us, Ordering::Relaxed);

                                let _ = req.reply.send(PhpResponse::Ok {
                                    status: req_ctx.status_code,
                                    body: req_ctx.output_buf,
                                    headers: req_ctx.response_headers,
                                    exec_time_us,
                                });
                            }
                            Err(_) => {
                                // On timeout, we intentionally do not free ctx_ptr to avoid double-free with the PHP worker thread. The worker will clean up when it completes.
                                let _ = req
                                    .reply
                                    .send(PhpResponse::Error("PHP worker timeout".into()));
                                dispatcher_total_errors.fetch_add(1, Ordering::Relaxed);
                            }
                        }

                        dispatcher_active.fetch_sub(1, Ordering::Relaxed);
                    }
                }
            }
        });

    let dispatcher = match dispatcher {
        Ok(handle) => Some(handle),
        Err(e) => {
            tracing::error!(worker = worker_id, error = %e, "Failed to spawn dispatcher thread");
            return;
        }
    };

    // Run the PHP worker script on this thread.
    // It will loop in bext_handle_request(), blocking on request_rx.
    let exit_status = unsafe {
        ffi::bext_php_execute_worker(
            &mut boot_ctx as *mut RequestCtx as *mut ffi::BextRequestCtx,
            c_script.as_ptr(),
        )
    };

    tracing::info!(worker = worker_id, exit_status, "PHP worker script exited");

    // Clean up: drop request_tx to unblock the dispatcher, then join it
    drop(request_tx);
    if let Some(d) = dispatcher {
        let _ = d.join();
    }

    // Clean up thread-local state
    WORKER_THREAD_STATE.with(|state| {
        *state.borrow_mut() = None;
    });
}

// ─── Classic mode execution ──────────────────────────────────────────────

#[allow(clippy::too_many_arguments)]
fn execute_classic(
    script_path: &str,
    method: &str,
    uri: &str,
    query_string: &str,
    content_type: Option<&str>,
    body: Vec<u8>,
    cookies: Option<&str>,
    headers: Vec<(String, String)>,
    remote_addr: Option<&str>,
    server_name: Option<&str>,
    server_port: u16,
    https: bool,
) -> PhpResponse {
    let c_script = match CString::new(script_path) {
        Ok(s) => s,
        Err(e) => return PhpResponse::Error(format!("Invalid script path: {}", e)),
    };
    let c_method = match CString::new(method) {
        Ok(s) => s,
        Err(e) => return PhpResponse::Error(format!("Invalid method: {}", e)),
    };
    let c_uri = match CString::new(uri) {
        Ok(s) => s,
        Err(e) => return PhpResponse::Error(format!("Invalid URI: {}", e)),
    };
    let c_query = match CString::new(query_string) {
        Ok(s) => s,
        Err(e) => return PhpResponse::Error(format!("Invalid query string: {}", e)),
    };
    let c_content_type = content_type.and_then(|ct| CString::new(ct).ok());
    let content_length = body.len() as i64;

    let mut req_ctx = RequestCtx::new(
        body,
        cookies,
        headers,
        remote_addr,
        server_name,
        server_port,
        https,
    );

    let start = Instant::now();

    let status = unsafe {
        ffi::bext_php_execute_script(
            &mut req_ctx as *mut RequestCtx as *mut ffi::BextRequestCtx,
            c_script.as_ptr(),
            c_method.as_ptr(),
            c_uri.as_ptr(),
            c_query.as_ptr(),
            c_content_type
                .as_ref()
                .map(|c| c.as_ptr())
                .unwrap_or(std::ptr::null()),
            content_length,
        )
    };

    let exec_time_us = start.elapsed().as_micros() as u64;

    if status < 0 {
        return PhpResponse::Error("PHP execution failed (request startup error)".into());
    }

    PhpResponse::Ok {
        status: status as u16,
        body: req_ctx.output_buf,
        headers: req_ctx.response_headers,
        exec_time_us,
    }
}