Opensource internals

Polars Runs Queries on Rayon's Global Thread Pool

13 min read Jay Hubbert

Polars owns almost no threads: query work runs on rayon's global pool, async orchestration on a global Tokio runtime, and blocking escapes to Tokio's blocking pool.

TL;DR: Polars does not have a thread pool of its own for query execution. crates/polars-core/src/runtime.rs defines a zero-sized RAYON marker whose install, join, scope and spawn forward to rayon’s free functions — which, on a native build, means rayon’s global pool. Polars’ own ThreadPools are a single-threaded pool for munmap and a wasm-only pool. Async orchestration lives in a separate global Tokio runtime (ASYNC), and RAYON::block_on is the escape hatch that keeps a blocking call from parking a rayon worker.

I went looking for the pool where Polars runs its queries, expecting something like a POOL global with a per-engine lifecycle. What is actually there is more interesting, and slightly the opposite of what I expected: on a native build Polars does not create a thread pool at all for query execution. It borrows rayon’s.

One marker type for every parallel entry point

The whole abstraction lives in crates/polars-core/src/runtime.rs, and it starts with a type that has no fields:

rust 
pub struct RAYON;

// Thread locals to allow disabling threading for specific threads.
#[cfg(any(target_os = "emscripten", not(target_family = "wasm")))]
thread_local! {
    static NOOP_POOL: RefCell<ThreadPool> = RefCell::new(
        ThreadPoolBuilder::new()
            .use_current_thread()
            .num_threads(1)
            .build()
            .expect("could not create no-op thread pool")
    );
}

crates/polars-core/src/runtime.rs:13

RAYON is a zero-sized type, so RAYON as a value is a singleton that carries no state. Its methods are the only way the rest of Polars is allowed to touch rayon, and every one of them has two implementations: a native one and a wasm one.

rust 
impl RAYON {
    pub fn install<OP, R>(&self, op: OP) -> R
    where
        OP: FnOnce() -> R + Send,
        R: Send,
    {
        #[cfg(not(any(target_os = "emscripten", not(target_family = "wasm"))))]
        {
            op()
        }

        #[cfg(any(target_os = "emscripten", not(target_family = "wasm")))]
        {
            self.with(|p| p.install(op))
        }
    }

    pub fn join<A, B, RA, RB>(&self, oper_a: A, oper_b: B) -> (RA, RB)
    where
        A: FnOnce() -> RA + Send,
        B: FnOnce() -> RB + Send,
        RA: Send,
        RB: Send,
    {
        self.install(|| rayon::join(oper_a, oper_b))
    }

crates/polars-core/src/runtime.rs:39

Look at the native branch of install: it is literally op(). No pool, no ThreadPool::install, nothing. The rayon call happens inside op, and because rayon parallel iterators (par_iter) use the current pool — the global one when you are not inside an explicit install — the work lands on rayon’s global thread pool.

spawn and spawn_fifo are the same story:

rust 
    pub fn spawn<OP>(&self, op: OP)
    where
        OP: FnOnce() + Send + 'static,
    {
        #[cfg(not(any(target_os = "emscripten", not(target_family = "wasm"))))]
        {
            rayon::spawn(op)
        }

        #[cfg(any(target_os = "emscripten", not(target_family = "wasm")))]
        {
            self.with(|p| {
                p.spawn(op);
                if p.current_num_threads() == 1 {
                    p.yield_now();
                }
            })
        }
    }

crates/polars-core/src/runtime.rs:70

This is a deliberate design decision, and it has an obvious motivation: a library should not decide how many threads the host application gets. If Polars spun up its own pool of n_cores threads, then a Python process that also uses numpy’s threads, or a Rust application that already has a rayon pool, would end up with two competing thread pools fighting over the same cores. Sharing rayon’s global pool means there is exactly one pool of worker threads in the process, and Polars is a guest in it.

The cost is that Polars cannot bound its own parallelism independently. The pool size is rayon’s, which is POLARS_MAX_THREADS if Polars built the pool first (see below) and otherwise whatever the host application configured.

The pools Polars does own

There are exactly two ThreadPoolBuilders in the whole workspace outside of tests. The first is the wasm-only pool:

rust 
// this is re-exported in utils for polars child crates
#[cfg(not(target_family = "wasm"))] // only use this on non wasm targets
pub static THREAD_POOL: LazyLock<ThreadPool> = LazyLock::new(|| {
    let thread_name = std::env::var("POLARS_THREAD_NAME").unwrap_or_else(|_| "polars".to_string());
    ThreadPoolBuilder::new()
        .num_threads(polars_config::config().max_threads())
        .thread_name(move |i| format!("{thread_name}-{i}"))
        .build()
        .expect("could not spawn threads")
});

crates/polars-core/src/runtime.rs:194

Read the cfg carefully: on native builds this pool exists, but it is never installed — nothing runs on it — and it is only consulted by RAYON::block_on to ask whether the current thread is one of its workers. The method that does install it, RAYON::with, is compiled only for emscripten/wasm:

rust 
    #[cfg(any(target_os = "emscripten", not(target_family = "wasm")))]
    pub fn with<OP, R>(&self, op: OP) -> R
    where
        OP: FnOnce(&ThreadPool) -> R + Send,
        R: Send,
    {
        if polars_async::executor::ALLOW_RAYON_THREADS.get()
            || THREAD_POOL.current_thread_index().is_some()
        {
            op(&THREAD_POOL)
        } else {
            NOOP_POOL.with(|v| op(&v.borrow()))
        }
    }

crates/polars-core/src/runtime.rs:139

On wasm there is no shared address space to parallelise over and no real threading, so Polars uses a one-thread pool that runs the work on the current thread, and falls back to the NOOP_POOL when it must not run in parallel at all. This is the one place where the “own pool” design is real, and it exists precisely because the global-pool trick is not available.

The second pool is the UNMAP_POOL in crates/polars-utils/src/mmap.rs, with exactly one thread, and it exists for a reason that has nothing to do with queries:

rust 
pub static UNMAP_POOL: LazyLock<ThreadPool> = LazyLock::new(|| {
    let thread_name = std::env::var("POLARS_THREAD_NAME").unwrap_or_else(|_| "polars".to_string());
    ThreadPoolBuilder::new()
        .num_threads(1)
        .thread_name(move |i| format!("{thread_name}-unmap-{i}"))
        .build()
        .expect("could not spawn threads")
});

crates/polars-utils/src/mmap.rs:16

munmap takes a lock on the file being unmapped, and unmapping a large mapping can block for a long time. Doing that on a rayon worker would stall the whole pool, so Polars moves it to a dedicated single-threaded pool. This is the pattern to notice: Polars does not create pools for throughput, it creates pools for isolation.

Never block a rayon thread

The same reasoning produces the most interesting function in the file. Rayon workers are supposed to be doing work; parking one on a mutex or a channel receive steals a core from every other job. RAYON::block_on is the escape hatch, and it moves the blocking part onto Tokio’s blocking pool:

rust 
    /// Calls a blocking function without blocking the rayon thread pool by
    /// moving it to a different thread.
    ///
    /// If this thread isn't a rayon thread this simply calls f directly.
    pub fn block_on<R: Send, F: FnOnce() -> R + Send>(&self, f: F) -> R {
        if THREAD_POOL.current_thread_index().is_some() {
            // We wrap F such that the resulting closure no longer has F or R as
            // parameters, and whose only limitation is that it isn't 'static.
            let mut opt_f: Option<F> = Some(f);
            let mut out: Option<std::thread::Result<R>> = None;
            let mut wrap_f = || {
                let f = AssertUnwindSafe(opt_f.take().unwrap());
                out = Some(catch_unwind(f));
            };

            // SAFETY: we always await the future to completion before returning from here, meaning
            // wrap_f stays alive for as long as it needs to. If for whatever reason we unwind we
            // abort.
            let abort = WithDrop::new((), |()| std::process::abort);

crates/polars-core/src/runtime.rs:152

Two details worth pausing on. First, the WithDrop::new((), |std::process::abort) guard: if the future is dropped before it completes, the closure that owns F would be dropped on the wrong thread, so the process is aborted instead of risking UB. Second, catch_unwind around the moved closure: a panic in the blocking work has to travel back across the thread boundary.

The caller then spins, yielding to rayon rather than blocking it:

rust 
            while !join_handle.is_finished() {
                match rayon::yield_now() {
                    Some(Yield::Executed) => {},
                    Some(Yield::Idle) => std::thread::yield_now(),
                    None => unreachable!(),
                }
            }

crates/polars-core/src/runtime.rs:175

rayon::yield_now lets the worker run other queued jobs while it waits, which is the whole point: the rayon thread is not idle, it is doing someone else’s work.

Note the pool the guard asks about: THREAD_POOL.current_thread_index(), i.e. the pool Polars builds for itself, not rayon’s global pool. That distinction is worth keeping in mind, because it is the seam between the two worlds: the blocking escape hatch is written for threads that belong to a pool Polars controls, because a thread from an arbitrary pool has no business being parked in ASYNC.spawn_blocking and then unwound through catch_unwind on somebody else’s scheduler.

The async runtime

Everything above is synchronous. The orchestration layer — the streaming engine’s graph of operators, the parquet reader’s I/O — lives on a Tokio runtime, and it is a global one:

rust 
pub struct RuntimeManager {
    rt: Runtime,
}

impl RuntimeManager {
    fn new() -> Self {
        let n_threads = std::env::var("POLARS_ASYNC_THREAD_COUNT")
            .map(|x| x.parse::<usize>().expect("integer"))
            .unwrap_or(usize::min(polars_config::config().max_threads(), 32));

        let max_blocking = std::env::var("POLARS_MAX_BLOCKING_THREAD_COUNT")
            .map(|x| x.parse::<usize>().expect("integer"))
            .unwrap_or(512);

crates/polars-async/src/lib.rs:23

rust 
        let rt = Builder::new_multi_thread()
            .worker_threads(n_threads)
            .max_blocking_threads(max_blocking)
            .on_thread_start(move || {
                if tokio_thread_count_start.fetch_add(1) + 1 >= (max_total_threads as i64) && !warned.load() {
                    warned.store(true);
                    polars_warn!("POLARS_MAX_BLOCKING_THREAD_COUNT reached ({max_blocking}), this may indicate a deadlock");
                }
            })

crates/polars-async/src/lib.rs:37

The warning is worth noting: the blocking pool is the one RAYON::block_on and spawn_blocking feed from, and if it fills up, something is waiting on a blocking call that will never complete — a deadlock waiting to happen. The runtime warns rather than hangs.

The manager is a process global:

rust 
/// The global Polars async runtime accessor.
pub static ASYNC: LazyLock<RuntimeManager> = LazyLock::new(RuntimeManager::new);

crates/polars-async/src/lib.rs:120

and polars-core re-exports it next to RAYON:

rust 
pub use polars_async::ASYNC;

crates/polars-core/src/runtime.rs:211

So the two halves of the parallelism story sit next to each other in one file: rayon for data-parallel CPU work, Tokio for orchestration and I/O.

How the in-memory engine uses it

The in-memory engine’s executors are ordinary rayon users, routed through the marker. The filter executor is a good example because it has to decide whether splitting the frame is even worth it:

rust 
    /// Evaluates the predicate, in parallel over row chunks if worthwhile.
    fn evaluate_predicate(&self, df: &DataFrame, state: &ExecutionState) -> PolarsResult<Column> {
        let n_chunks =
            (df.height() / MIN_ROWS_PER_PREDICATE_CHUNK).min(RAYON.current_num_threads());
        let is_column = matches!(self.predicate.as_expression(), Some(Expr::Column(_)));
        if !self.streamable || n_chunks < 2 || is_column {
            return self.predicate.evaluate(df, state);
        }

        let chunks = df.clone().split_chunks_by_n(n_chunks, true);
        let columns = RAYON.install(|| {
            chunks
                .into_par_iter()
                .map(|chunk| self.predicate.evaluate(&chunk, state))
                .collect::<PolarsResult<Vec<_>>>()
        })?;

crates/polars-mem-engine/src/executors/filter.rs:67

RAYON.current_num_threads() is the budget: the frame is split into at most as many chunks as there are workers, and only if there are at least two. The group-by executor does the same with a parallel iterator over aggregations:

rust 
pub(super) fn evaluate_aggs(
    df: &DataFrame,
    aggs: &[Arc<dyn PhysicalExpr>],
    groups: &GroupPositions,
    state: &ExecutionState,
) -> PolarsResult<Vec<Column>> {
    RAYON.install(|| {
        aggs.par_iter()
            .map(|expr| {
                let agg = expr.evaluate_on_groups(df, groups, state)?.finalize();
                polars_ensure!(agg.len() == groups.len(), agg_len = agg.len(), groups.len());
                Ok(agg)
            })
            .collect::<PolarsResult<Vec<_>>>()
    })
}

crates/polars-mem-engine/src/executors/group_by.rs:15

Note that RAYON.install is called around the par_iter, not around each element. On native that is a no-op wrapper; on wasm it is what pins the parallel iterator to Polars’ own single-thread pool.

The streaming engine mixes both worlds. The join runtime filter, for instance, fans a morsel batch over rayon and then blocks on an async key selection inside each chunk:

rust 
        let chunk_len = morsels.len().div_ceil(RAYON.current_num_threads()).max(1);
        let locals = RAYON.install(|| {
            morsels
                .par_chunks(chunk_len)
                .map(|chunk| {
                    let mut builders = self.new_builders();
                    for morsel in chunk {
                        let df = morsel.df_blocking();
                        let keys = ASYNC.block_on(select_key_columns(&df, key_selectors, state))?;
                        self.extend(&keys, &mut builders)?;
                    }
                    PolarsResult::Ok(builders)
                })
                .collect::<PolarsResult<Vec<_>>>()
        })?;

crates/polars-stream/src/nodes/joins/runtime_filter.rs:100

ASYNC.block_on inside a rayon worker is precisely the case RAYON::block_on was written for — but here it is the Tokio runtime’s own block_on, which runs the future to completion on the current thread. It is safe because select_key_columns is genuinely async and the thread is going to be busy either way.

For a query that must not block the caller at all, the lazy engine hands the whole thing to a rayon worker and returns a handle:

rust 
        } else {
            RAYON.spawn_fifo(move || {
                let result = physical_plan.execute(&mut state);
                tx.send(result).unwrap();
            });
        }

crates/polars-lazy/src/frame/exitable.rs:20

spawn_fifo rather than spawn: FIFO scheduling means the query’s tasks are run in submission order, which keeps memory behaviour predictable for a pipeline that stages data.

Errors across task boundaries

Two small utilities in crates/polars-utils/src/async_utils/ handle the part that is easy to get wrong when work is spread across tasks: making sure an error in one task is not silently swallowed by the others.

error_capture.rs gives every task an ErrorCapture and the parent an ErrorHandle:

rust 
    /// Wraps a future such that its error result is sent to the associated [`ErrorHandle`].
    pub async fn wrap_future<F, O>(self, fut: F)
    where
        F: Future<Output = Result<O, ErrorT>>,
    {
        let err: Result<(), tokio::sync::mpsc::error::TrySendError<ErrorMessage<ErrorT>>> =
            match AssertUnwindSafe(fut).catch_unwind().await {
                Ok(Ok(_)) => return,
                Ok(Err(err)) => self.tx.try_send(ErrorMessage::Error(err)),
                Err(panic) => self.tx.try_send(ErrorMessage::Panic(panic)),
            };
        drop(err);
    }

crates/polars-utils/src/async_utils/error_capture.rs:30

The catch_unwind is what turns a panic in a spawned task into a message the parent can act on, and ErrorHandle::join re-raises it:

rust 
    /// Block until either an error is received, or all [`ErrorCapture`]s associated with this
    /// handle are dropped (i.e. successful completion of all wrapped futures).
    ///
    /// # Panics
    /// If a panic is received, this will resume unwinding.
    pub async fn join(self) -> Result<(), ErrorT> {
        let ErrorHandle { mut rx } = self;

        match rx.recv().await {
            None => Ok(()),
            Some(ErrorMessage::Error(e)) => Err(e),
            Some(ErrorMessage::Panic(panic)) => std::panic::resume_unwind(panic),
        }
    }

crates/polars-utils/src/async_utils/error_capture.rs:67

And tokio_handle_ext.rs is the abort-on-drop guard, which matters because a cancelled query must not leave a task running against a half-torn-down plan:

rust 
/// Calls [`tokio::task::JoinHandle::abort`] on the join handle when dropped.
pub struct AbortOnDropHandle<T>(pub tokio::task::JoinHandle<T>);

impl<T> Future for AbortOnDropHandle<T> {
    type Output = Result<T, tokio::task::JoinError>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        Pin::new(&mut self.0).poll(cx)
    }
}

impl<T> Drop for AbortOnDropHandle<T> {
    fn drop(&mut self) {
        self.0.abort();
    }
}

crates/polars-utils/src/async_utils/tokio_handle_ext.rs:11

Send, Sync, and the one unsafe escape

Since Polars hands work to threads it does not own, the Send/Sync bounds matter a lot, and there is exactly one place where Polars asserts them manually. crates/polars-utils/src/sync.rs is not a general-purpose concurrency toolkit; it holds a single type:

rust 
/// Utility that allows use to send pointers to another thread.
/// This is better than going through `usize` as MIRI can follow these.
#[derive(Debug, PartialEq, Eq)]
#[repr(transparent)]
pub struct SyncPtr<T>(*mut T);

crates/polars-utils/src/sync.rs:5

The unsafe impl Sync for SyncPtr<T> and unsafe impl Send for SyncPtr<T> at the bottom of the file are the whole point: a raw pointer is neither, and this wrapper launders it into something that can cross a thread boundary while staying visible to Miri. The deref_unchecked accessor is correspondingly unsafe and documents that there are no guarantees at all beyond the pointer being valid.

Everything else in Polars that crosses threads does it with ordinary Arc/Send/Sync, and the Unsync-style opt-outs that older versions of Polars used to mark types as thread-local do not exist in this tree.

The bottom line

  • Polars has no thread pool for query execution on native builds. crates/polars-core/src/runtime.rs routes every parallel operation through the RAYON marker, whose native branch forwards to rayon’s global pool. One pool per process, not one per library.
  • The pools Polars does own are for isolation, not throughput: a one-thread UNMAP_POOL so munmap cannot stall a worker, and a wasm-only THREAD_POOL for platforms with no shared-memory parallelism.
  • Blocking is banned on rayon workers. RAYON::block_on moves the blocking part to Tokio’s blocking pool and yields the rayon thread to other jobs meanwhile, aborting the process rather than risking a cross-thread drop if the future is cancelled.
  • Orchestration is Tokio, and it is global too. ASYNC is a multi-thread runtime with POLARS_ASYNC_THREAD_COUNT workers (default: min(cores, 32)) and POLARS_MAX_BLOCKING_THREAD_COUNT blocking threads (default 512), and it warns when the blocking pool fills up because that is the shape of a deadlock.
  • The number of pipelines is Polars’ own. The streaming engine builds polars_config::config().max_threads() pipelines (crates/polars-stream/src/execute.rs:293), so POLARS_MAX_THREADS — which defaults to available_parallelism() — controls how many concurrent operators a streaming query runs, even though the workers underneath come from rayon’s shared pool.

For a user, the practical knobs are therefore POLARS_MAX_THREADS (pipelines and Tokio workers) plus the Tokio thread counts, and the practical rule is: never write a Python UDF or a plugin that blocks, because it will be occupying a shared worker that other queries need.

PolarsInternalsConcurrencyPerformance