Skip to main content

async_ebpf/
program.rs

1//! This module contains the core logic for eBPF program execution. A lot of
2//! `unsafe` code is used - be careful when making changes here.
3
4use std::{
5  any::{Any, TypeId},
6  cell::{Cell, RefCell},
7  collections::HashMap,
8  ffi::CStr,
9  marker::PhantomData,
10  mem::ManuallyDrop,
11  ops::{Deref, DerefMut},
12  pin::Pin,
13  ptr::NonNull,
14  rc::Rc,
15  sync::{
16    atomic::{compiler_fence, AtomicBool, AtomicU64, Ordering},
17    Arc, Once,
18  },
19  task::{Context, Poll},
20  thread::ThreadId,
21  time::{Duration, Instant},
22};
23
24use corosensei::{
25  stack::{DefaultStack, Stack},
26  Coroutine, CoroutineResult, ScopedCoroutine, Yielder,
27};
28use futures::{task::noop_waker_ref, Future, FutureExt};
29use memmap2::{MmapOptions, MmapRaw};
30use parking_lot::{Condvar, Mutex};
31use rand::prelude::SliceRandom;
32
33use crate::{
34  error::{Error, RuntimeError},
35  function_analysis::{analyze_functions, FunctionLayout},
36  helpers::Helper,
37  linker::{link_elf, validate_local_call_graph},
38  pointer_cage::PointerCage,
39  region_analysis::PointerSignature,
40  util::nonnull_bytes_overlap,
41};
42
43const NATIVE_STACK_SIZE: usize = 16384;
44const SHADOW_STACK_SIZE: usize = 32768;
45const MAX_CALLDATA_SIZE: usize = 512;
46const MAX_MUTABLE_DEREF_REGIONS: usize = 4;
47const MAX_IMMUTABLE_DEREF_REGIONS: usize = 16;
48
49#[cfg(all(target_arch = "x86_64", target_os = "linux"))]
50std::arch::global_asm!(
51  r#"
52.global async_ebpf_entry_trampoline
53.type async_ebpf_entry_trampoline,@function
54async_ebpf_entry_trampoline:
55    mov r10, rdi
56    mov rax, [rsp + 8]
57    push rbp
58    push rbx
59    push r12
60    push r13
61    push r14
62    push r15
63    mov r15, rcx
64    add r15, r8
65    mov r11, rcx
66    mov rdi, rsi
67    mov rsi, rdx
68    mov rdx, r11
69    mov rcx, r8
70    mov r8, r9
71    mov r11, rdi
72    sub rsp, 8
73    mov rbp, rsp
74    sub rsp, 32
75    mov [rbp - 8], rax
76    call r10
77    mov rsp, rbp
78    add rsp, 8
79    pop r15
80    pop r14
81    pop r13
82    pop r12
83    pop rbx
84    pop rbp
85    ret
86.size async_ebpf_entry_trampoline, . - async_ebpf_entry_trampoline
87"#
88);
89
90#[cfg(all(target_arch = "aarch64", target_os = "linux"))]
91std::arch::global_asm!(
92  r#"
93.global async_ebpf_entry_trampoline
94.type async_ebpf_entry_trampoline,%function
95async_ebpf_entry_trampoline:
96    mov x17, x0
97    sub sp, sp, #16
98    stp x29, x30, [sp]
99    sub sp, sp, #64
100    stp x19, x20, [sp, #0]
101    stp x21, x22, [sp, #16]
102    stp x23, x24, [sp, #32]
103    stp x25, x26, [sp, #48]
104    mov x29, sp
105    add x23, x3, x4
106    mov x0, x1
107    mov x1, x2
108    mov x2, x3
109    mov x3, x4
110    mov x4, x5
111    sub sp, sp, #16
112    str x6, [x29, #-8]
113    mov x26, x0
114    blr x17
115    mov x0, x5
116    mov sp, x29
117    ldp x19, x20, [sp, #0]
118    ldp x21, x22, [sp, #16]
119    ldp x23, x24, [sp, #32]
120    ldp x25, x26, [sp, #48]
121    add sp, sp, #64
122    ldp x29, x30, [sp]
123    add sp, sp, #16
124    ret
125.size async_ebpf_entry_trampoline, . - async_ebpf_entry_trampoline
126"#
127);
128
129extern "C" {
130  fn async_ebpf_entry_trampoline(
131    target: usize,
132    ctx: usize,
133    mem_len: usize,
134    stack: usize,
135    stack_len: usize,
136    reserved: usize,
137    memory: usize,
138  ) -> u64;
139}
140
141/// Per-invocation storage for helper state during a program run.
142pub struct InvokeScope {
143  data: HashMap<TypeId, Box<dyn Any + Send>>,
144}
145
146impl InvokeScope {
147  /// Gets or creates typed data scoped to this invocation.
148  pub fn data_mut<T: Default + Send + 'static>(&mut self) -> &mut T {
149    let ty = TypeId::of::<T>();
150    self
151      .data
152      .entry(ty)
153      .or_insert_with(|| Box::new(T::default()))
154      .downcast_mut()
155      .expect("InvokeScope::data_mut: downcast failed")
156  }
157}
158
159/// Context passed to helpers while a program is executing.
160pub struct HelperScope<'a, 'b> {
161  /// The program being executed.
162  pub program: &'a Program,
163  /// Mutable per-invocation data for helpers.
164  pub invoke: RefCell<&'a mut InvokeScope>,
165  resources: RefCell<&'a mut [&'b mut dyn Any]>,
166  memory: &'a JitMemory,
167  mutable_dereferenced_regions: [Cell<Option<NonNull<[u8]>>>; MAX_MUTABLE_DEREF_REGIONS],
168  immutable_dereferenced_regions: [Cell<Option<NonNull<[u8]>>>; MAX_IMMUTABLE_DEREF_REGIONS],
169  can_post_task: bool,
170}
171
172/// A validated mutable view into user memory.
173pub struct MutableUserMemory<'a, 'b, 'c> {
174  _scope: &'c HelperScope<'a, 'b>,
175  region: NonNull<[u8]>,
176}
177
178impl<'a, 'b, 'c> Deref for MutableUserMemory<'a, 'b, 'c> {
179  type Target = [u8];
180
181  fn deref(&self) -> &Self::Target {
182    unsafe { self.region.as_ref() }
183  }
184}
185
186impl<'a, 'b, 'c> DerefMut for MutableUserMemory<'a, 'b, 'c> {
187  fn deref_mut(&mut self) -> &mut Self::Target {
188    unsafe { self.region.as_mut() }
189  }
190}
191
192impl<'a, 'b> HelperScope<'a, 'b> {
193  /// Posts an async task to be run between timeslices.
194  pub fn post_task(
195    &self,
196    task: impl Future<Output = impl FnOnce(&HelperScope) -> Result<u64, ()> + 'static> + 'static,
197  ) {
198    if !self.can_post_task {
199      panic!("HelperScope::post_task() called in a context where posting task is not allowed");
200    }
201
202    PENDING_ASYNC_TASK.with(|x| {
203      let mut x = x.borrow_mut();
204      if x.is_some() {
205        panic!("post_task called while another task is pending");
206      }
207      *x = Some(async move { Box::new(task.await) as AsyncTaskOutput }.boxed_local());
208    });
209  }
210
211  /// Calls `callback` with a mutable resource of type `T`, if present.
212  pub fn with_resource_mut<'c, T: 'static, R>(
213    &'c self,
214    callback: impl FnOnce(Result<&mut T, ()>) -> R,
215  ) -> R {
216    let mut resources = self.resources.borrow_mut();
217    let Some(res) = resources
218      .iter_mut()
219      .filter_map(|x| x.downcast_mut::<T>())
220      .next()
221    else {
222      tracing::warn!(resource_type = ?TypeId::of::<T>(), "resource not found");
223      return callback(Err(()));
224    };
225
226    callback(Ok(res))
227  }
228
229  /// Validates and returns an immutable view into user memory.
230  pub fn user_memory(&self, ptr: u64, size: u64) -> Result<&[u8], ()> {
231    let Some(region) = self.memory.safe_deref_for_read(ptr as usize, size as usize) else {
232      tracing::warn!(ptr, size, "invalid read");
233      return Err(());
234    };
235
236    if size != 0 {
237      // The region must not overlap with any previously dereferenced mutable regions
238      if self
239        .mutable_dereferenced_regions
240        .iter()
241        .filter_map(|x| x.get())
242        .any(|x| nonnull_bytes_overlap(x, region))
243      {
244        tracing::warn!(ptr, size, "read overlapped with previous write");
245        return Err(());
246      }
247
248      // Find a slot to record this dereference
249      let Some(slot) = self
250        .immutable_dereferenced_regions
251        .iter()
252        .find(|x| x.get().is_none())
253      else {
254        tracing::warn!(ptr, size, "too many reads");
255        return Err(());
256      };
257      slot.set(Some(region));
258    }
259
260    Ok(unsafe { region.as_ref() })
261  }
262
263  /// Validates and returns a mutable view into user memory.
264  pub fn user_memory_mut<'c>(
265    &'c self,
266    ptr: u64,
267    size: u64,
268  ) -> Result<MutableUserMemory<'a, 'b, 'c>, ()> {
269    let Some(region) = self
270      .memory
271      .safe_deref_for_write(ptr as usize, size as usize)
272    else {
273      tracing::warn!(ptr, size, "invalid write");
274      return Err(());
275    };
276
277    if size != 0 {
278      // The region must not overlap with any other previously dereferenced mutable or immutable regions
279      if self
280        .mutable_dereferenced_regions
281        .iter()
282        .chain(self.immutable_dereferenced_regions.iter())
283        .filter_map(|x| x.get())
284        .any(|x| nonnull_bytes_overlap(x, region))
285      {
286        tracing::warn!(ptr, size, "write overlapped with previous read/write");
287        return Err(());
288      }
289
290      // Find a slot to record this dereference
291      let Some(slot) = self
292        .mutable_dereferenced_regions
293        .iter()
294        .find(|x| x.get().is_none())
295      else {
296        tracing::warn!(ptr, size, "too many writes");
297        return Err(());
298      };
299      slot.set(Some(region));
300    }
301
302    Ok(MutableUserMemory {
303      _scope: self,
304      region,
305    })
306  }
307}
308
309#[derive(Copy, Clone)]
310struct AssumeSend<T>(T);
311unsafe impl<T> Send for AssumeSend<T> {}
312
313struct ExecContext {
314  native_stack: DefaultStack,
315  guest_stack: Box<[u8; SHADOW_STACK_SIZE]>,
316}
317
318impl ExecContext {
319  fn new() -> Self {
320    Self {
321      native_stack: DefaultStack::new(NATIVE_STACK_SIZE)
322        .expect("failed to initialize native stack"),
323      guest_stack: Box::new([0u8; SHADOW_STACK_SIZE]),
324    }
325  }
326}
327
328#[repr(C)]
329struct JitMemory {
330  stack_guest_bottom: usize,
331  stack_guest_top: usize,
332  stack_native_base: usize,
333  data_guest_bottom: usize,
334  data_guest_top: usize,
335  data_native_base: usize,
336}
337
338impl JitMemory {
339  fn checked_region(
340    guest: usize,
341    size: usize,
342    guest_bottom: usize,
343    guest_top: usize,
344    native_base: usize,
345  ) -> Option<NonNull<[u8]>> {
346    if size == 0 {
347      return Some(NonNull::slice_from_raw_parts(NonNull::dangling(), 0));
348    }
349
350    let end = guest.checked_add(size)?;
351    if guest < guest_bottom || end > guest_top {
352      return None;
353    }
354    let native = native_base.checked_add(guest - guest_bottom)? as *mut u8;
355    unsafe {
356      Some(NonNull::new_unchecked(std::ptr::slice_from_raw_parts_mut(
357        native, size,
358      )))
359    }
360  }
361
362  fn safe_deref_for_write(&self, guest: usize, size: usize) -> Option<NonNull<[u8]>> {
363    Self::checked_region(
364      guest,
365      size,
366      self.stack_guest_bottom,
367      self.stack_guest_top,
368      self.stack_native_base,
369    )
370  }
371
372  fn safe_deref_for_read(&self, guest: usize, size: usize) -> Option<NonNull<[u8]>> {
373    Self::checked_region(
374      guest,
375      size,
376      self.stack_guest_bottom,
377      self.stack_guest_top,
378      self.stack_native_base,
379    )
380    .or_else(|| {
381      Self::checked_region(
382        guest,
383        size,
384        self.data_guest_bottom,
385        self.data_guest_top,
386        self.data_native_base,
387      )
388    })
389  }
390}
391
392/// A pending async task spawned by a helper.
393pub type PendingAsyncTask = Pin<Box<dyn Future<Output = AsyncTaskOutput>>>;
394/// The callback produced by a helper async task when it resumes.
395pub type AsyncTaskOutput = Box<dyn FnOnce(&HelperScope) -> Result<u64, ()>>;
396
397static NEXT_PROGRAM_ID: AtomicU64 = AtomicU64::new(1);
398
399#[derive(Copy, Clone, Debug)]
400enum PreemptionState {
401  Inactive,
402  Armed(usize),
403  Shutdown,
404}
405
406type PreemptionStateSignal = (Mutex<PreemptionState>, Condvar);
407
408thread_local! {
409  static RUST_TID: ThreadId = std::thread::current().id();
410  static SIGUSR1_COUNTER: Cell< u64> = Cell::new(0);
411  static ACTIVE_JIT_CODE_ZONE: ActiveJitCodeZone = ActiveJitCodeZone::default();
412  static EXEC_CONTEXT_POOL: RefCell<Vec<ExecContext>> = Default::default();
413  static PENDING_ASYNC_TASK: RefCell<Option<PendingAsyncTask>> = RefCell::new(None);
414  static PREEMPTION_STATE: Arc<PreemptionStateSignal> = Arc::new((Mutex::new(PreemptionState::Inactive), Condvar::new()));
415  static LOADING_PROGRAM_LOADER: Cell<*const ProgramLoader> = const { Cell::new(std::ptr::null()) };
416  static ACTIVE_PROGRAM: Cell<*const Program> = const { Cell::new(std::ptr::null()) };
417}
418
419struct BorrowedExecContext {
420  ctx: ManuallyDrop<ExecContext>,
421}
422
423impl BorrowedExecContext {
424  fn new() -> Self {
425    let mut me = Self {
426      ctx: ManuallyDrop::new(
427        EXEC_CONTEXT_POOL.with(|x| x.borrow_mut().pop().unwrap_or_else(ExecContext::new)),
428      ),
429    };
430    me.ctx.guest_stack.fill(0x8e);
431    me
432  }
433}
434
435impl Drop for BorrowedExecContext {
436  fn drop(&mut self) {
437    let ctx = unsafe { ManuallyDrop::take(&mut self.ctx) };
438    EXEC_CONTEXT_POOL.with(|x| x.borrow_mut().push(ctx));
439  }
440}
441
442#[derive(Default)]
443struct ActiveJitCodeZone {
444  valid: AtomicBool,
445  code_range: Cell<(usize, usize)>,
446  pointer_cage_protected_range: Cell<(usize, usize)>,
447  yielder: Cell<Option<NonNull<Yielder<u64, Dispatch>>>>,
448}
449
450/// Hooks for observing program execution events.
451pub trait ProgramEventListener: Send + Sync + 'static {
452  /// Called after an async preemption is triggered.
453  fn did_async_preempt(&self, _scope: &HelperScope) {}
454  /// Called after yielding back to the async runtime.
455  fn did_yield(&self) {}
456  /// Called after throttling a program's execution.
457  fn did_throttle(&self, _scope: &HelperScope) -> Option<Pin<Box<dyn Future<Output = ()>>>> {
458    None
459  }
460}
461
462/// No-op event listener implementation.
463pub struct DummyProgramEventListener;
464impl ProgramEventListener for DummyProgramEventListener {}
465
466/// Default limit for the total JIT-compiled native code size of one program.
467pub const DEFAULT_CODE_SIZE_LIMIT: usize = 1 << 20;
468
469/// Prepares helper tables and loads eBPF programs.
470pub struct ProgramLoader {
471  helpers_inverse: HashMap<&'static str, i32>,
472  event_listener: Arc<dyn ProgramEventListener>,
473  helper_id_xor: u16,
474  helpers: Arc<Vec<(u16, &'static str, Helper)>>,
475  code_size_limit: usize,
476  require_static_regions: bool,
477}
478
479/// A loaded program that is not yet pinned to a thread.
480pub struct UnboundProgram {
481  id: u64,
482  _code_mem: MmapRaw,
483  code_base: usize,
484  code_size: usize,
485  page_size: usize,
486  code_arena: RefCell<CodeArena>,
487  cage: PointerCage,
488  helper_id_xor: u16,
489  helpers: Arc<Vec<(u16, &'static str, Helper)>>,
490  event_listener: Arc<dyn ProgramEventListener>,
491  require_static_regions: bool,
492  entrypoints: HashMap<String, usize>,
493  sections: RefCell<Vec<Section>>,
494  resolvers: RefCell<HashMap<u32, ResolverInfo>>,
495  next_resolver_id: Cell<u32>,
496}
497
498/// A program pinned to a specific thread and ready to execute.
499pub struct Program {
500  unbound: UnboundProgram,
501  data: RefCell<HashMap<TypeId, Rc<dyn Any>>>,
502  t: ThreadEnv,
503}
504
505#[derive(Copy, Clone)]
506struct Entrypoint {
507  code_ptr: usize,
508}
509
510struct CodeArena {
511  used: usize,
512}
513
514struct Section {
515  vm: Vm,
516  code_vaddr: usize,
517  code_len: usize,
518  layout: FunctionLayout,
519  functions: Vec<FunctionState>,
520}
521
522#[derive(Default)]
523struct FunctionState {
524  compiled: HashMap<PointerSignature, FunctionCompilation>,
525  #[cfg(test)]
526  compile_attempts: usize,
527}
528
529#[derive(Clone)]
530enum FunctionCompilation {
531  Succeeded(Entrypoint),
532  Failed(RuntimeError),
533}
534
535impl FunctionCompilation {
536  fn result(&self) -> Result<Entrypoint, RuntimeError> {
537    match self {
538      Self::Succeeded(entrypoint) => Ok(*entrypoint),
539      Self::Failed(err) => Err(err.clone()),
540    }
541  }
542
543  fn entrypoint(&self) -> Option<Entrypoint> {
544    match self {
545      Self::Succeeded(entrypoint) => Some(*entrypoint),
546      Self::Failed(_) => None,
547    }
548  }
549}
550
551#[derive(Clone, Copy)]
552struct ResolverInfo {
553  section_index: usize,
554  function_index: usize,
555  signature: PointerSignature,
556}
557
558/// Time limits used to yield or throttle execution.
559#[derive(Clone, Debug)]
560pub struct TimesliceConfig {
561  /// Maximum runtime before yielding to the async scheduler.
562  pub max_run_time_before_yield: Duration,
563  /// Maximum runtime before a throttle sleep is forced.
564  pub max_run_time_before_throttle: Duration,
565  /// Duration of the throttle sleep once triggered.
566  pub throttle_duration: Duration,
567}
568
569/// Async runtime integration for yielding and sleeping.
570pub trait Timeslicer {
571  /// Sleep for the provided duration.
572  fn sleep(&self, duration: Duration) -> impl Future<Output = ()>;
573  /// Yield to the async scheduler.
574  fn yield_now(&self) -> impl Future<Output = ()>;
575}
576
577/// Global runtime environment for signal handlers.
578#[derive(Copy, Clone)]
579pub struct GlobalEnv(());
580
581/// Per-thread runtime environment for preemption handling.
582#[derive(Copy, Clone)]
583pub struct ThreadEnv {
584  _not_send_sync: std::marker::PhantomData<*const ()>,
585}
586
587impl GlobalEnv {
588  /// Initializes global state and installs signal handlers.
589  ///
590  /// # Safety
591  /// Must be called in a process that can install SIGUSR1/SIGSEGV handlers.
592  pub unsafe fn new() -> Self {
593    static INIT: Once = Once::new();
594
595    // SIGUSR1 must be blocked during exception handling
596    // Otherwise it seems that Linux gives up and throws an uncatchable SI_KERNEL SIGSEGV:
597    //
598    // [pid 517110] tgkill(517109, 517112, SIGUSR1 <unfinished ...>
599    // [pid 517112] --- SIGSEGV {si_signo=SIGSEGV, si_code=SEGV_ACCERR, si_addr=0x793789be227e} ---
600    // [pid 517109] write(15, "\1\0\0\0\0\0\0\0", 8 <unfinished ...>
601    // [pid 517110] <... tgkill resumed>)      = 0
602    // [pid 517109] <... write resumed>)       = 8
603    // [pid 517112] --- SIGUSR1 {si_signo=SIGUSR1, si_code=SI_TKILL, si_pid=517109, si_uid=1000} ---
604    // [pid 517109] recvfrom(57,  <unfinished ...>
605    // [pid 517110] futex(0x79378abff5a8, FUTEX_WAIT_PRIVATE, 1, {tv_sec=0, tv_nsec=5999909} <unfinished ...>
606    // [pid 517109] <... recvfrom resumed>"GET /write_rodata HTTP/1.1\r\nHost"..., 8192, 0, NULL, NULL) = 61
607    // [pid 517112] --- SIGSEGV {si_signo=SIGSEGV, si_code=SI_KERNEL, si_addr=NULL} ---
608
609    INIT.call_once(|| {
610      let sa_mask = get_blocked_sigset();
611
612      for (sig, handler) in [
613        (libc::SIGUSR1, sigusr1_handler as *const () as usize),
614        (libc::SIGSEGV, sigsegv_handler as *const () as usize),
615      ] {
616        let act = libc::sigaction {
617          sa_sigaction: handler,
618          sa_flags: libc::SA_SIGINFO,
619          sa_mask,
620          sa_restorer: None,
621        };
622        if libc::sigaction(sig, &act, std::ptr::null_mut()) != 0 {
623          panic!("failed to setup handler for signal {}", sig);
624        }
625      }
626    });
627
628    Self(())
629  }
630
631  /// Initializes per-thread state and starts the async preemption watcher.
632  pub fn init_thread(self, async_preemption_interval: Duration) -> ThreadEnv {
633    struct DeferDrop(Arc<PreemptionStateSignal>);
634    impl Drop for DeferDrop {
635      fn drop(&mut self) {
636        let x = &self.0;
637        *x.0.lock() = PreemptionState::Shutdown;
638        x.1.notify_one();
639      }
640    }
641
642    thread_local! {
643      static WATCHER: RefCell<Option<DeferDrop>> = RefCell::new(None);
644    }
645
646    if WATCHER.with(|x| x.borrow().is_some()) {
647      return ThreadEnv {
648        _not_send_sync: PhantomData,
649      };
650    }
651
652    let preemption_state = PREEMPTION_STATE.with(|x| x.clone());
653
654    unsafe {
655      let tgid = libc::getpid();
656      let tid = libc::gettid();
657
658      std::thread::Builder::new()
659        .name("preempt-watcher".to_string())
660        .spawn(move || {
661          let mut state = preemption_state.0.lock();
662          loop {
663            match *state {
664              PreemptionState::Shutdown => break,
665              PreemptionState::Inactive => {
666                preemption_state.1.wait(&mut state);
667              }
668              PreemptionState::Armed(_) => {
669                let timeout = preemption_state.1.wait_while_for(
670                  &mut state,
671                  |x| matches!(x, PreemptionState::Armed(_)),
672                  async_preemption_interval,
673                );
674                if timeout.timed_out() {
675                  match *state {
676                    PreemptionState::Armed(0) => {
677                      *state = PreemptionState::Inactive;
678                    }
679                    PreemptionState::Armed(_) => {
680                      let ret = libc::syscall(libc::SYS_tgkill, tgid, tid, libc::SIGUSR1);
681                      if ret != 0 {
682                        break;
683                      }
684                    }
685                    PreemptionState::Inactive => {}
686                    PreemptionState::Shutdown => break,
687                  }
688                }
689              }
690            }
691          }
692        })
693        .expect("failed to spawn preemption watcher");
694
695      WATCHER.with(|x| {
696        x.borrow_mut()
697          .replace(DeferDrop(PREEMPTION_STATE.with(|x| x.clone())));
698      });
699
700      ThreadEnv {
701        _not_send_sync: PhantomData,
702      }
703    }
704  }
705}
706
707impl UnboundProgram {
708  /// Pins the program to the current thread using a prepared `ThreadEnv`.
709  pub fn pin_to_current_thread(self, t: ThreadEnv) -> Program {
710    Program {
711      unbound: self,
712      data: RefCell::new(HashMap::new()),
713      t,
714    }
715  }
716}
717
718pub struct PreemptionEnabled(());
719
720impl PreemptionEnabled {
721  pub fn new(_: ThreadEnv) -> Self {
722    PREEMPTION_STATE.with(|x| {
723      let mut notify = false;
724      {
725        let mut st = x.0.lock();
726        let next = match *st {
727          PreemptionState::Inactive => {
728            notify = true;
729            PreemptionState::Armed(1)
730          }
731          PreemptionState::Armed(n) => PreemptionState::Armed(n + 1),
732          PreemptionState::Shutdown => unreachable!(),
733        };
734        *st = next;
735      }
736
737      if notify {
738        x.1.notify_one();
739      }
740    });
741    Self(())
742  }
743}
744
745impl Drop for PreemptionEnabled {
746  fn drop(&mut self) {
747    PREEMPTION_STATE.with(|x| {
748      let mut st = x.0.lock();
749      let next = match *st {
750        PreemptionState::Armed(1) => PreemptionState::Armed(0),
751        PreemptionState::Armed(n) => {
752          assert!(n > 1);
753          PreemptionState::Armed(n - 1)
754        }
755        PreemptionState::Inactive | PreemptionState::Shutdown => unreachable!(),
756      };
757      *st = next;
758    });
759  }
760}
761
762impl Program {
763  /// Returns the unique program identifier.
764  pub fn id(&self) -> u64 {
765    self.unbound.id
766  }
767
768  pub fn thread_env(&self) -> ThreadEnv {
769    self.t
770  }
771
772  /// Gets or creates shared typed data for this program instance.
773  pub fn data<T: Default + 'static>(&self) -> Rc<T> {
774    let mut data = self.data.borrow_mut();
775    let entry = data.entry(TypeId::of::<T>());
776    let entry = entry.or_insert_with(|| Rc::new(T::default()));
777    entry.clone().downcast().unwrap()
778  }
779
780  pub fn has_section(&self, name: &str) -> bool {
781    self.unbound.entrypoints.contains_key(name)
782  }
783
784  #[cfg(test)]
785  pub(crate) fn compiled_function_count_for_tests(&self) -> usize {
786    self
787      .unbound
788      .sections
789      .borrow()
790      .iter()
791      .map(|section| {
792        section
793          .functions
794          .iter()
795          .map(|function| {
796            function
797              .compiled
798              .values()
799              .filter(|compilation| compilation.entrypoint().is_some())
800              .count()
801          })
802          .sum::<usize>()
803      })
804      .sum()
805  }
806
807  #[cfg(test)]
808  pub(crate) fn failed_function_count_for_tests(&self) -> usize {
809    self
810      .unbound
811      .sections
812      .borrow()
813      .iter()
814      .map(|section| {
815        section
816          .functions
817          .iter()
818          .map(|function| {
819            function
820              .compiled
821              .values()
822              .filter(|compilation| compilation.entrypoint().is_none())
823              .count()
824          })
825          .sum::<usize>()
826      })
827      .sum()
828  }
829
830  #[cfg(test)]
831  pub(crate) fn function_compile_attempt_count_for_tests(&self) -> usize {
832    self
833      .unbound
834      .sections
835      .borrow()
836      .iter()
837      .map(|section| {
838        section
839          .functions
840          .iter()
841          .map(|function| function.compile_attempts)
842          .sum::<usize>()
843      })
844      .sum()
845  }
846
847  #[cfg(test)]
848  pub(crate) fn code_arena_used_for_tests(&self) -> usize {
849    self.unbound.code_arena.borrow().used
850  }
851
852  /// Number of successfully compiled variants per source function, flattened
853  /// across sections. One entry per function in the layout; a value > 1 means
854  /// that callee was specialized for multiple incoming pointer signatures.
855  #[cfg(test)]
856  pub(crate) fn function_variant_counts_for_tests(&self) -> Vec<usize> {
857    self
858      .unbound
859      .sections
860      .borrow()
861      .iter()
862      .flat_map(|section| {
863        section.functions.iter().map(|function| {
864          function
865            .compiled
866            .values()
867            .filter(|compilation| compilation.entrypoint().is_some())
868            .count()
869        })
870      })
871      .collect()
872  }
873
874  fn compile_entrypoint(&self, section_index: usize) -> Result<Entrypoint, RuntimeError> {
875    self.compile_function(section_index, 0, PointerSignature::entry())
876  }
877
878  fn compile_resolver(&self, resolver_id: u32) -> Result<Entrypoint, RuntimeError> {
879    let Some(info) = self.unbound.resolvers.borrow().get(&resolver_id).copied() else {
880      return Err(RuntimeError::InvalidArgument(
881        "local call resolver not found",
882      ));
883    };
884    self.compile_function(info.section_index, info.function_index, info.signature)
885  }
886
887  fn cached_resolver_target(&self, resolver_id: u32) -> Option<usize> {
888    let info = self.unbound.resolvers.borrow().get(&resolver_id).copied()?;
889    let sections = self.unbound.sections.borrow();
890    let section = sections.get(info.section_index)?;
891    section
892      .functions
893      .get(info.function_index)?
894      .compiled
895      .get(&info.signature)
896      .and_then(|compilation| compilation.entrypoint())
897      .map(|entrypoint| entrypoint.code_ptr)
898  }
899
900  fn protect_code_pages(&self, executable_len: usize) -> Result<(), RuntimeError> {
901    let executable_len =
902      (executable_len + self.unbound.page_size - 1) & !(self.unbound.page_size - 1);
903    unsafe {
904      if executable_len > 0
905        && libc::mprotect(
906          self.unbound.code_base as *mut _,
907          executable_len,
908          libc::PROT_READ | libc::PROT_EXEC,
909        ) != 0
910      {
911        return Err(RuntimeError::PlatformError(
912          "failed to protect executable code",
913        ));
914      }
915      if executable_len < self.unbound.code_size
916        && libc::mprotect(
917          (self.unbound.code_base + executable_len) as *mut _,
918          self.unbound.code_size - executable_len,
919          libc::PROT_NONE,
920        ) != 0
921      {
922        return Err(RuntimeError::PlatformError("failed to protect unused code"));
923      }
924    }
925    Ok(())
926  }
927
928  fn compile_function(
929    &self,
930    section_index: usize,
931    function_index: usize,
932    signature: PointerSignature,
933  ) -> Result<Entrypoint, RuntimeError> {
934    {
935      let sections = self.unbound.sections.borrow();
936      let Some(section) = sections.get(section_index) else {
937        return Err(RuntimeError::InvalidArgument("section not found"));
938      };
939      let Some(function) = section.functions.get(function_index) else {
940        return Err(RuntimeError::InvalidArgument("function not found"));
941      };
942      if let Some(compilation) = function.compiled.get(&signature) {
943        return compilation.result();
944      }
945    }
946
947    let mut sections = self.unbound.sections.borrow_mut();
948    let section = sections
949      .get_mut(section_index)
950      .ok_or(RuntimeError::InvalidArgument("section not found"))?;
951    if function_index >= section.functions.len() {
952      return Err(RuntimeError::InvalidArgument("function not found"));
953    }
954    if let Some(compilation) = section.functions[function_index].compiled.get(&signature) {
955      return compilation.result();
956    }
957    #[cfg(test)]
958    {
959      section.functions[function_index].compile_attempts += 1;
960    }
961    let function = section.layout.functions[function_index].clone();
962    let code = self
963      .unbound
964      .cage
965      .safe_deref_for_read(section.code_vaddr, section.code_len)
966      .unwrap();
967    let code_bytes = unsafe { std::slice::from_raw_parts(code.as_ptr() as *const u8, code.len()) };
968    let region_analysis = crate::region_analysis::analyze_function(
969      code_bytes,
970      function.start_pc,
971      function.end_pc,
972      signature,
973      self.unbound.cage.data_bottom() as u64,
974      self.unbound.cage.data_top() as u64,
975    );
976    if self.unbound.require_static_regions && !region_analysis.unresolved.is_empty() {
977      let err = RuntimeError::InvalidArgumentOwned(format!(
978        "static region analysis failed in function [{}, {}): {} memory access(es) could not be \
979         routed to a single region (instruction slots {:?})",
980        function.start_pc,
981        function.end_pc,
982        region_analysis.unresolved.len(),
983        region_analysis.unresolved,
984      ));
985      section.functions[function_index]
986        .compiled
987        .insert(signature, FunctionCompilation::Failed(err.clone()));
988      return Err(err);
989    }
990
991    // Allocate resolver ids and build their metadata locally. Nothing is
992    // committed to `next_resolver_id` or the shared `resolvers` map until the
993    // function has been fully compiled and protected, so a failed compilation
994    // does not leak resolver ids or orphan map entries.
995    let mut resolver_ids = vec![0u32; code_bytes.len() / 8];
996    let mut pending_resolvers: Vec<(u32, ResolverInfo)> = Vec::new();
997    let mut next_resolver_id = self.unbound.next_resolver_id.get();
998    for (&call_pc, &callee_signature) in &region_analysis.call_signatures {
999      let target_pc = local_call_target(code_bytes, call_pc);
1000      let callee_index = section.layout.pc_to_func[target_pc];
1001      let resolver_id = next_resolver_id;
1002      let Some(advanced) = resolver_id.checked_add(1) else {
1003        let err = RuntimeError::InvalidArgument("too many local call resolvers");
1004        section.functions[function_index]
1005          .compiled
1006          .insert(signature, FunctionCompilation::Failed(err.clone()));
1007        return Err(err);
1008      };
1009      next_resolver_id = advanced;
1010      resolver_ids[call_pc] = resolver_id;
1011      pending_resolvers.push((
1012        resolver_id,
1013        ResolverInfo {
1014          section_index,
1015          function_index: callee_index,
1016          signature: callee_signature,
1017        },
1018      ));
1019    }
1020
1021    let mut arena = self.unbound.code_arena.borrow_mut();
1022    if arena.used >= self.unbound.code_size {
1023      let err = RuntimeError::InvalidArgument("no space left for jit compilation");
1024      section.functions[function_index]
1025        .compiled
1026        .insert(signature, FunctionCompilation::Failed(err.clone()));
1027      return Err(err);
1028    }
1029
1030    unsafe {
1031      if libc::mprotect(
1032        self.unbound.code_base as *mut _,
1033        self.unbound.code_size,
1034        libc::PROT_READ | libc::PROT_WRITE,
1035      ) != 0
1036      {
1037        let err = RuntimeError::PlatformError("failed to make code writable");
1038        section.functions[function_index]
1039          .compiled
1040          .insert(signature, FunctionCompilation::Failed(err.clone()));
1041        return Err(err);
1042      }
1043    }
1044
1045    let code_ptr = self.unbound.code_base + arena.used;
1046    let mut written_len = self.unbound.code_size - arena.used;
1047    let mut errmsg_ptr = std::ptr::null_mut();
1048
1049    let ret = unsafe {
1050      crate::ubpf::ubpf_set_region_hints(
1051        section.vm.0.as_ptr(),
1052        region_analysis.hints.as_ptr(),
1053        region_analysis.hints.len(),
1054      );
1055      crate::ubpf::ubpf_set_lazy_local_call_resolver(
1056        section.vm.0.as_ptr(),
1057        Some(tls_local_call_resolver),
1058        resolver_ids.as_ptr(),
1059        resolver_ids.len(),
1060      );
1061      let ret = crate::ubpf::ubpf_translate_function_ex(
1062        section.vm.0.as_ptr(),
1063        code_ptr as *mut u8,
1064        &mut written_len,
1065        &mut errmsg_ptr,
1066        crate::ubpf::JitMode_ExtendedJitMode,
1067        function.start_pc as u32,
1068        function.end_pc as u32,
1069      );
1070      crate::ubpf::ubpf_set_region_hints(section.vm.0.as_ptr(), std::ptr::null(), 0);
1071      crate::ubpf::ubpf_set_lazy_local_call_resolver(
1072        section.vm.0.as_ptr(),
1073        None,
1074        std::ptr::null(),
1075        0,
1076      );
1077      ret
1078    };
1079
1080    if ret != 0 {
1081      let errmsg = unsafe {
1082        if errmsg_ptr.is_null() {
1083          "".to_string()
1084        } else {
1085          CStr::from_ptr(errmsg_ptr).to_string_lossy().into_owned()
1086        }
1087      };
1088      if !errmsg_ptr.is_null() {
1089        unsafe { libc::free(errmsg_ptr as _) };
1090      }
1091      let _ = self.protect_code_pages(arena.used);
1092      let err =
1093        RuntimeError::InvalidArgumentOwned(format!("ubpf: code translation failed: {errmsg}"));
1094      section.functions[function_index]
1095        .compiled
1096        .insert(signature, FunctionCompilation::Failed(err.clone()));
1097      return Err(err);
1098    }
1099    if !errmsg_ptr.is_null() {
1100      unsafe { libc::free(errmsg_ptr as _) };
1101    }
1102
1103    unsafe {
1104      crate::ubpf::ubpf_clear_instruction_cache(code_ptr as *mut u8, written_len);
1105    }
1106
1107    // Restore W^X protection covering the newly emitted function before
1108    // advancing the arena. If protection cannot be restored the function is not
1109    // executable, so leave `arena.used` unchanged (reclaiming the space, which
1110    // the next compilation overwrites after making the region writable again)
1111    // and do not register its resolvers.
1112    let new_used = arena.used + written_len;
1113    if let Err(err) = self.protect_code_pages(new_used) {
1114      section.functions[function_index]
1115        .compiled
1116        .insert(signature, FunctionCompilation::Failed(err.clone()));
1117      return Err(err);
1118    }
1119    arena.used = new_used;
1120
1121    // Compilation succeeded: commit the resolver ids and metadata.
1122    self.unbound.next_resolver_id.set(next_resolver_id);
1123    {
1124      let mut resolvers = self.unbound.resolvers.borrow_mut();
1125      for (resolver_id, info) in pending_resolvers {
1126        resolvers.insert(resolver_id, info);
1127      }
1128    }
1129
1130    let entrypoint = Entrypoint { code_ptr };
1131    section.functions[function_index]
1132      .compiled
1133      .insert(signature, FunctionCompilation::Succeeded(entrypoint));
1134    tracing::debug!(
1135      section_index,
1136      function_index,
1137      start_pc = function.start_pc,
1138      end_pc = function.end_pc,
1139      native_code_addr = ?(code_ptr as *const u8),
1140      native_code_size = written_len,
1141      "jit compiled function"
1142    );
1143    Ok(entrypoint)
1144  }
1145
1146  /// Runs the program entrypoint with the provided resources and calldata.
1147  pub async fn run(
1148    &self,
1149    timeslice: &TimesliceConfig,
1150    timeslicer: &impl Timeslicer,
1151    entrypoint: &str,
1152    resources: &mut [&mut dyn Any],
1153    calldata: &[u8],
1154    preemption: &PreemptionEnabled,
1155  ) -> Result<i64, Error> {
1156    self
1157      ._run(
1158        timeslice, timeslicer, entrypoint, resources, calldata, preemption,
1159      )
1160      .await
1161      .map_err(Error)
1162  }
1163
1164  async fn _run(
1165    &self,
1166    timeslice: &TimesliceConfig,
1167    timeslicer: &impl Timeslicer,
1168    entrypoint: &str,
1169    resources: &mut [&mut dyn Any],
1170    calldata: &[u8],
1171    _: &PreemptionEnabled,
1172  ) -> Result<i64, RuntimeError> {
1173    let Some(section_index) = self.unbound.entrypoints.get(entrypoint).copied() else {
1174      return Err(RuntimeError::InvalidArgument("entrypoint not found"));
1175    };
1176    let entrypoint = self.compile_entrypoint(section_index)?;
1177    struct CoDropper<'a, Input, Yield, Return, DefaultStack: Stack>(
1178      ScopedCoroutine<'a, Input, Yield, Return, DefaultStack>,
1179    );
1180    impl<'a, Input, Yield, Return, DefaultStack: Stack> Drop
1181      for CoDropper<'a, Input, Yield, Return, DefaultStack>
1182    {
1183      fn drop(&mut self) {
1184        // Prevent the coroutine library from attempting to unwind the stack of the coroutine
1185        // and run destructors, because this stack might be running a signal handler and
1186        // it's not allowed to unwind from there.
1187        //
1188        // SAFETY: The coroutine stack only contains stack frames for JIT-compiled code and
1189        // carefully chosen Rust code that do not hold Droppable values, so it's safe to
1190        // skip destructors.
1191        unsafe {
1192          self.0.force_reset();
1193        }
1194      }
1195    }
1196
1197    let mut ectx = BorrowedExecContext::new();
1198
1199    if calldata.len() > MAX_CALLDATA_SIZE {
1200      return Err(RuntimeError::InvalidArgument("calldata too large"));
1201    }
1202    ectx.ctx.guest_stack[SHADOW_STACK_SIZE - calldata.len()..].copy_from_slice(calldata);
1203    let calldata_len = calldata.len();
1204
1205    let program_ret: u64 = {
1206      let guest_stack_top = self.unbound.cage.stack_top();
1207      let guest_stack_bottom = self.unbound.cage.stack_bottom();
1208      let ctx = &mut *ectx.ctx;
1209      let memory = JitMemory {
1210        stack_guest_bottom: guest_stack_bottom,
1211        stack_guest_top: guest_stack_top,
1212        stack_native_base: ctx.guest_stack.as_mut_ptr() as usize,
1213        data_guest_bottom: self.unbound.cage.data_bottom(),
1214        data_guest_top: self.unbound.cage.data_top(),
1215        data_native_base: self.unbound.cage.data_native_base(),
1216      };
1217      let memory_ptr = &memory as *const JitMemory as usize;
1218
1219      let mut co = AssumeSend(CoDropper(Coroutine::with_stack(
1220        &mut ctx.native_stack,
1221        move |yielder, _input| unsafe {
1222          ACTIVE_JIT_CODE_ZONE.with(|x| {
1223            x.yielder.set(NonNull::new(yielder as *const _ as *mut _));
1224          });
1225          let calldata_start = guest_stack_top - calldata_len;
1226          let stack_top = calldata_start & !0x7;
1227          let stack_len = stack_top - guest_stack_bottom;
1228          async_ebpf_entry_trampoline(
1229            entrypoint.code_ptr,
1230            calldata_start,
1231            calldata_start,
1232            guest_stack_bottom,
1233            stack_len,
1234            0,
1235            memory_ptr,
1236          )
1237        },
1238      )));
1239
1240      let mut last_yield_time: Option<Instant> = None;
1241      let mut last_throttle_time: Option<Instant> = None;
1242      let mut yielder: Option<AssumeSend<NonNull<Yielder<u64, Dispatch>>>> = None;
1243      let mut resume_input: u64 = 0;
1244      let mut did_throttle = false;
1245      let mut rust_tid_sigusr1_counter = (RUST_TID.with(|x| *x), SIGUSR1_COUNTER.with(|x| x.get()));
1246      let mut prev_async_task_output: Option<(&'static str, AsyncTaskOutput)> = None;
1247      let mut invoke_scope = InvokeScope {
1248        data: HashMap::new(),
1249      };
1250
1251      loop {
1252        ACTIVE_JIT_CODE_ZONE.with(|x| {
1253          x.code_range.set((
1254            self.unbound.code_base,
1255            self.unbound.code_base + self.unbound.code_size,
1256          ));
1257          x.yielder.set(yielder.map(|x| x.0));
1258          x.pointer_cage_protected_range.set((0, 4096));
1259          compiler_fence(Ordering::Release);
1260          x.valid.store(true, Ordering::Relaxed);
1261        });
1262        ACTIVE_PROGRAM.with(|x| x.set(self as *const _));
1263
1264        // If the previous iteration wants to write back to machine state
1265        if let Some((helper_name, prev_async_task_output)) = prev_async_task_output.take() {
1266          resume_input = prev_async_task_output(&HelperScope {
1267            program: self,
1268            invoke: RefCell::new(&mut invoke_scope),
1269            resources: RefCell::new(resources),
1270            memory: &memory,
1271            mutable_dereferenced_regions: unsafe { std::mem::zeroed() },
1272            immutable_dereferenced_regions: unsafe { std::mem::zeroed() },
1273            can_post_task: false,
1274          })
1275          .map_err(|_| RuntimeError::AsyncHelperError(helper_name))?;
1276        }
1277
1278        let ret = co.0 .0.resume(resume_input);
1279        ACTIVE_PROGRAM.with(|x| x.set(std::ptr::null()));
1280        ACTIVE_JIT_CODE_ZONE.with(|x| {
1281          x.valid.store(false, Ordering::Relaxed);
1282          compiler_fence(Ordering::Release);
1283          yielder = x.yielder.get().map(AssumeSend);
1284        });
1285
1286        let dispatch: Dispatch = match ret {
1287          CoroutineResult::Return(x) => break x,
1288          CoroutineResult::Yield(x) => x,
1289        };
1290
1291        // restore signal mask of current thread
1292        if dispatch.memory_access_error.is_some() || dispatch.async_preemption {
1293          unsafe {
1294            let unblock = get_blocked_sigset();
1295            libc::sigprocmask(libc::SIG_UNBLOCK, &unblock, std::ptr::null_mut());
1296          }
1297        }
1298
1299        if let Some(si_addr) = dispatch.memory_access_error {
1300          let vaddr = if si_addr >= memory.stack_native_base
1301            && si_addr < memory.stack_native_base + SHADOW_STACK_SIZE
1302          {
1303            memory.stack_guest_bottom + (si_addr - memory.stack_native_base)
1304          } else if si_addr >= memory.data_native_base
1305            && si_addr
1306              < memory.data_native_base + (memory.data_guest_top - memory.data_guest_bottom)
1307          {
1308            memory.data_guest_bottom + (si_addr - memory.data_native_base)
1309          } else {
1310            0
1311          };
1312          return Err(RuntimeError::MemoryFault(vaddr));
1313        }
1314
1315        if let Some(resolver_id) = dispatch.lazy_local_call {
1316          resume_input = self.compile_resolver(resolver_id)?.code_ptr as u64;
1317          continue;
1318        }
1319
1320        // Clear pending task if something else has set it
1321        PENDING_ASYNC_TASK.with(|x| x.borrow_mut().take());
1322        let mut helper_name: &'static str = "";
1323
1324        let mut helper_scope = HelperScope {
1325          program: self,
1326          invoke: RefCell::new(&mut invoke_scope),
1327          resources: RefCell::new(resources),
1328          memory: &memory,
1329          mutable_dereferenced_regions: unsafe { std::mem::zeroed() },
1330          immutable_dereferenced_regions: unsafe { std::mem::zeroed() },
1331          can_post_task: false,
1332        };
1333
1334        if dispatch.async_preemption {
1335          self
1336            .unbound
1337            .event_listener
1338            .did_async_preempt(&mut helper_scope);
1339        } else {
1340          // validator should ensure all helper indexes are present in the table
1341          let Some((_, got_helper_name, helper)) = self
1342            .unbound
1343            .helpers
1344            .get(
1345              ((dispatch.index & 0xffff) as u16 ^ self.unbound.helper_id_xor).wrapping_sub(1)
1346                as usize,
1347            )
1348            .copied()
1349          else {
1350            panic!("unknown helper index: {}", dispatch.index);
1351          };
1352          helper_name = got_helper_name;
1353
1354          helper_scope.can_post_task = true;
1355          resume_input = helper(
1356            &mut helper_scope,
1357            dispatch.arg1,
1358            dispatch.arg2,
1359            dispatch.arg3,
1360            dispatch.arg4,
1361            dispatch.arg5,
1362          )
1363          .map_err(|()| RuntimeError::HelperError(helper_name))?;
1364          helper_scope.can_post_task = false;
1365        }
1366
1367        let pending_async_task = PENDING_ASYNC_TASK.with(|x| x.borrow_mut().take());
1368
1369        // Fast path: do not read timestamp if no thread migration or async preemption happened
1370        let new_rust_tid_sigusr1_counter =
1371          (RUST_TID.with(|x| *x), SIGUSR1_COUNTER.with(|x| x.get()));
1372        if new_rust_tid_sigusr1_counter == rust_tid_sigusr1_counter && pending_async_task.is_none()
1373        {
1374          continue;
1375        }
1376
1377        rust_tid_sigusr1_counter = new_rust_tid_sigusr1_counter;
1378
1379        let now = Instant::now();
1380        let last_throttle = last_throttle_time.get_or_insert(now);
1381        let last_yield = last_yield_time.get_or_insert(now);
1382        let should_throttle = now > *last_throttle
1383          && now.duration_since(*last_throttle) >= timeslice.max_run_time_before_throttle;
1384        let should_yield = now > *last_yield
1385          && now.duration_since(*last_yield) >= timeslice.max_run_time_before_yield;
1386        if should_throttle || should_yield || pending_async_task.is_some() {
1387          // we are now free to give up control of current thread to other async tasks
1388
1389          if should_throttle {
1390            if !did_throttle {
1391              did_throttle = true;
1392              tracing::warn!("throttling program");
1393            }
1394            timeslicer.sleep(timeslice.throttle_duration).await;
1395            let now = Instant::now();
1396            last_throttle_time = Some(now);
1397            last_yield_time = Some(now);
1398            let task = self.unbound.event_listener.did_throttle(&mut helper_scope);
1399            if let Some(task) = task {
1400              task.await;
1401            }
1402          } else if should_yield {
1403            timeslicer.yield_now().await;
1404            let now = Instant::now();
1405            last_yield_time = Some(now);
1406            self.unbound.event_listener.did_yield();
1407          }
1408
1409          // Now we have released all exclusive resources and can safely execute the async task.
1410          if let Some(mut pending_async_task) = pending_async_task {
1411            // Fast path: helpers that merely compute a value (the common case)
1412            // complete on the first poll without ever suspending. Poll once and,
1413            // if it's ready, skip the `Instant::now()` reads used for run-budget
1414            // compensation below. This matters because `clock_gettime` can cost
1415            // ~200ns on virtualized hosts, and this path runs on every async
1416            // helper invocation.
1417            let output =
1418              match pending_async_task.poll_unpin(&mut Context::from_waker(noop_waker_ref())) {
1419                Poll::Ready(output) => output,
1420                Poll::Pending => {
1421                  // The task genuinely suspended; account for the wall-clock time
1422                  // it took so it isn't charged against the program's run budget.
1423                  let async_start = Instant::now();
1424                  let output = pending_async_task.await;
1425                  let async_dur = async_start.elapsed();
1426                  if let Some(last_throttle_time) = &mut last_throttle_time {
1427                    *last_throttle_time += async_dur;
1428                  }
1429                  if let Some(last_yield_time) = &mut last_yield_time {
1430                    *last_yield_time += async_dur;
1431                  }
1432                  output
1433                }
1434              };
1435            prev_async_task_output = Some((helper_name, output));
1436          }
1437        }
1438      }
1439    };
1440
1441    Ok(program_ret as i64)
1442  }
1443}
1444
1445struct Vm(NonNull<crate::ubpf::ubpf_vm>);
1446unsafe impl Send for Vm {}
1447
1448impl Vm {
1449  fn new(cage: &PointerCage) -> Self {
1450    let vm = NonNull::new(unsafe { crate::ubpf::ubpf_create() }).expect("failed to create ubpf_vm");
1451    unsafe {
1452      crate::ubpf::ubpf_toggle_bounds_check(vm.as_ptr(), false);
1453      crate::ubpf::ubpf_set_jit_pointer_mask_and_offset(vm.as_ptr(), cage.mask(), cage.offset());
1454    }
1455    Self(vm)
1456  }
1457}
1458
1459impl Drop for Vm {
1460  fn drop(&mut self) {
1461    unsafe {
1462      crate::ubpf::ubpf_destroy(self.0.as_ptr());
1463    }
1464  }
1465}
1466
1467struct LoaderValidationScope {
1468  previous: *const ProgramLoader,
1469}
1470
1471impl LoaderValidationScope {
1472  fn new(loader: &ProgramLoader) -> Self {
1473    let previous = LOADING_PROGRAM_LOADER.with(|x| {
1474      let previous = x.get();
1475      x.set(loader as *const _);
1476      previous
1477    });
1478    Self { previous }
1479  }
1480}
1481
1482impl Drop for LoaderValidationScope {
1483  fn drop(&mut self) {
1484    LOADING_PROGRAM_LOADER.with(|x| x.set(self.previous));
1485  }
1486}
1487
1488fn local_call_target(code: &[u8], pc: usize) -> usize {
1489  let offset = pc * 8;
1490  let imm = i32::from_le_bytes([
1491    code[offset + 4],
1492    code[offset + 5],
1493    code[offset + 6],
1494    code[offset + 7],
1495  ]);
1496  (pc as i64 + imm as i64 + 1) as usize
1497}
1498
1499impl ProgramLoader {
1500  /// Creates a new `ProgramLoader` to load eBPF code.
1501  pub fn new(
1502    rng: &mut impl rand::Rng,
1503    event_listener: Arc<dyn ProgramEventListener>,
1504    raw_helpers: &[&[(&'static str, Helper)]],
1505  ) -> Self {
1506    let helper_id_xor = rng.gen::<u16>();
1507    let mut helpers_inverse: HashMap<&'static str, i32> = HashMap::new();
1508    // Collect first to a HashMap then to a Vec to deduplicate
1509    let mut shuffled_helpers = raw_helpers
1510      .iter()
1511      .flat_map(|x| x.iter().copied())
1512      .collect::<HashMap<_, _>>()
1513      .into_iter()
1514      .collect::<Vec<_>>();
1515    shuffled_helpers.shuffle(rng);
1516    let mut helpers: Vec<(u16, &'static str, Helper)> = Vec::with_capacity(shuffled_helpers.len());
1517
1518    assert!(shuffled_helpers.len() <= 65535);
1519
1520    for (i, (name, helper)) in shuffled_helpers.into_iter().enumerate() {
1521      let entropy = rng.gen::<u16>() & 0x7fff;
1522      helpers.push((entropy, name, helper));
1523      helpers_inverse.insert(
1524        name,
1525        (((entropy as usize) << 16) | ((i + 1) ^ (helper_id_xor as usize))) as i32,
1526      );
1527    }
1528
1529    tracing::info!(?helpers_inverse, "generated helper table");
1530    Self {
1531      helper_id_xor,
1532      helpers: Arc::new(helpers),
1533      helpers_inverse,
1534      event_listener,
1535      code_size_limit: DEFAULT_CODE_SIZE_LIMIT,
1536      require_static_regions: false,
1537    }
1538  }
1539
1540  /// Requires every guest memory access to be statically routable to a single
1541  /// region (stack or read-only data). When enabled, compilation fails if the
1542  /// region analysis cannot classify any load, store, or atomic — i.e. no
1543  /// access falls back to the dual-region runtime probe. Memory accesses
1544  /// reached only through unmodeled control flow (e.g. an argument pointer in a
1545  /// local function) count as unresolved and are rejected.
1546  ///
1547  /// Because functions are JIT-compiled lazily and per pointer-signature
1548  /// specialization, this check runs when each function variant is first
1549  /// compiled — at the start of [`Program::run`], not at load time. The error
1550  /// is therefore surfaced through `Program::run`. A function variant that is
1551  /// never reached is never compiled and never checked; the guarantee is "every
1552  /// executed access is statically routable", scoped to the variants actually
1553  /// invoked, rather than a whole-section load-time guarantee. (Soundness does
1554  /// not depend on this flag: unclassified accesses still get the dual-region
1555  /// runtime probe, so the flag only tightens strictness.)
1556  pub fn require_static_region_analysis(mut self, require: bool) -> Self {
1557    self.require_static_regions = require;
1558    self
1559  }
1560
1561  /// Sets the maximum total size of JIT-compiled native code per program.
1562  ///
1563  /// Must be a non-zero multiple of 64 KiB so the code region stays
1564  /// page-aligned on all supported page sizes.
1565  ///
1566  /// Note for arm64: conditional branches and literal loads emitted by the
1567  /// JIT have a ±1 MiB range, and a section's helper calls reference a
1568  /// literal pool at the end of that section's code. Any single ELF section
1569  /// whose jitted code exceeds ~1 MiB is therefore rejected at load time
1570  /// ("ubpf: code translation failed") regardless of this limit; raising it
1571  /// past 1 MiB only adds room for more or larger sections within that
1572  /// per-section ceiling. x86-64 has no such per-section constraint.
1573  pub fn with_code_size_limit(mut self, limit: usize) -> Self {
1574    assert!(
1575      limit > 0 && limit % (64 * 1024) == 0,
1576      "code size limit must be a non-zero multiple of 64 KiB"
1577    );
1578    assert!(
1579      limit <= u32::MAX as usize,
1580      "code size limit must fit in u32"
1581    );
1582    self.code_size_limit = limit;
1583    self
1584  }
1585
1586  /// Loads an ELF image into a new `UnboundProgram`.
1587  pub fn load(&self, rng: &mut impl rand::Rng, elf: &[u8]) -> Result<UnboundProgram, Error> {
1588    self._load(rng, elf).map_err(Error)
1589  }
1590
1591  fn _load(&self, rng: &mut impl rand::Rng, elf: &[u8]) -> Result<UnboundProgram, RuntimeError> {
1592    let start_time = Instant::now();
1593    let cage = PointerCage::new(rng, SHADOW_STACK_SIZE, elf.len())?;
1594
1595    let code_sections = {
1596      let mut data = cage
1597        .safe_deref_for_read(cage.data_bottom(), elf.len())
1598        .unwrap();
1599      let data = unsafe { data.as_mut() };
1600      data.copy_from_slice(elf);
1601
1602      link_elf(data, cage.data_bottom(), &self.helpers_inverse).map_err(RuntimeError::Linker)?
1603    };
1604    cage.freeze_data();
1605
1606    let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
1607    if page_size < 0 {
1608      return Err(RuntimeError::PlatformError("failed to get page size"));
1609    }
1610    let page_size = page_size as usize;
1611
1612    let guard_size_before = rng.gen_range(16..128) * page_size;
1613    let guard_size_after = rng.gen_range(16..128) * page_size;
1614
1615    let code_len_allocated = self.code_size_limit;
1616    let code_mem = MmapRaw::from(
1617      MmapOptions::new()
1618        .len(code_len_allocated + guard_size_before + guard_size_after)
1619        .map_anon()
1620        .map_err(|_| RuntimeError::PlatformError("failed to allocate code memory"))?,
1621    );
1622    let code_base = code_mem.as_ptr() as usize + guard_size_before;
1623
1624    unsafe {
1625      if libc::mprotect(
1626        code_mem.as_mut_ptr() as *mut _,
1627        guard_size_before,
1628        libc::PROT_NONE,
1629      ) != 0
1630        || libc::mprotect(
1631          code_mem
1632            .as_mut_ptr()
1633            .offset((guard_size_before + code_len_allocated) as isize) as *mut _,
1634          guard_size_after,
1635          libc::PROT_NONE,
1636        ) != 0
1637        || libc::mprotect(
1638          code_mem.as_mut_ptr().offset(guard_size_before as isize) as *mut _,
1639          code_len_allocated,
1640          libc::PROT_NONE,
1641        ) != 0
1642      {
1643        return Err(RuntimeError::PlatformError("failed to protect code memory"));
1644      }
1645    }
1646
1647    let mut entrypoints = HashMap::new();
1648    let mut sections = Vec::new();
1649    let resolvers = HashMap::new();
1650    let next_resolver_id = 1u32;
1651
1652    for (section_name, code_vaddr_size) in code_sections {
1653      let vm = Vm::new(&cage);
1654      unsafe {
1655        if crate::ubpf::ubpf_register_external_dispatcher(
1656          vm.0.as_ptr(),
1657          Some(tls_dispatcher),
1658          Some(std_validator),
1659        ) != 0
1660        {
1661          return Err(RuntimeError::PlatformError(
1662            "ubpf: failed to register external dispatcher",
1663          ));
1664        }
1665      }
1666
1667      let mut errmsg_ptr = std::ptr::null_mut();
1668      let code = cage
1669        .safe_deref_for_read(code_vaddr_size.0, code_vaddr_size.1)
1670        .unwrap();
1671      let code_bytes =
1672        unsafe { std::slice::from_raw_parts(code.as_ptr() as *const u8, code.len()) };
1673      validate_local_call_graph(code_bytes).map_err(|err| {
1674        RuntimeError::InvalidArgumentOwned(format!(
1675          "local call graph validation failed in {section_name}: {err}"
1676        ))
1677      })?;
1678      let layout = analyze_functions(code_bytes).map_err(|err| {
1679        RuntimeError::InvalidArgumentOwned(format!(
1680          "local function analysis failed in {section_name}: {err}"
1681        ))
1682      })?;
1683      let ret = unsafe {
1684        let validation_scope = LoaderValidationScope::new(self);
1685        let ret = crate::ubpf::ubpf_load(
1686          vm.0.as_ptr(),
1687          code.as_ptr() as *const _,
1688          code.len() as u32,
1689          &mut errmsg_ptr,
1690        );
1691        drop(validation_scope);
1692        ret
1693      };
1694      if ret != 0 {
1695        let errmsg = unsafe {
1696          if errmsg_ptr.is_null() {
1697            "".to_string()
1698          } else {
1699            CStr::from_ptr(errmsg_ptr).to_string_lossy().into_owned()
1700          }
1701        };
1702        if !errmsg_ptr.is_null() {
1703          unsafe { libc::free(errmsg_ptr as _) };
1704        }
1705        tracing::error!(section_name, error = errmsg, "failed to load code");
1706        return Err(RuntimeError::InvalidArgumentOwned(format!(
1707          "ubpf: code load failed: {errmsg}"
1708        )));
1709      }
1710
1711      let section_index = sections.len();
1712
1713      entrypoints.insert(section_name, section_index);
1714      let functions = (0..layout.functions.len())
1715        .map(|_| FunctionState::default())
1716        .collect();
1717      sections.push(Section {
1718        vm,
1719        code_vaddr: code_vaddr_size.0,
1720        code_len: code_vaddr_size.1,
1721        layout,
1722        functions,
1723      });
1724    }
1725
1726    tracing::info!(
1727      elf_size = elf.len(),
1728      native_code_addr = ?code_mem.as_ptr(),
1729      native_code_size_limit = code_len_allocated,
1730      guard_size_before,
1731      guard_size_after,
1732      duration = ?start_time.elapsed(),
1733      cage_ptr = ?cage.region().as_ptr(),
1734      cage_mapped_size = cage.region().len(),
1735      "loaded program for lazy jit"
1736    );
1737
1738    Ok(UnboundProgram {
1739      id: NEXT_PROGRAM_ID.fetch_add(1, Ordering::Relaxed),
1740      _code_mem: code_mem,
1741      code_base,
1742      code_size: code_len_allocated,
1743      page_size,
1744      code_arena: RefCell::new(CodeArena { used: 0 }),
1745      cage,
1746      helper_id_xor: self.helper_id_xor,
1747      helpers: self.helpers.clone(),
1748      event_listener: self.event_listener.clone(),
1749      require_static_regions: self.require_static_regions,
1750      entrypoints,
1751      sections: RefCell::new(sections),
1752      resolvers: RefCell::new(resolvers),
1753      next_resolver_id: Cell::new(next_resolver_id),
1754    })
1755  }
1756}
1757
1758#[derive(Default)]
1759struct Dispatch {
1760  async_preemption: bool,
1761  memory_access_error: Option<usize>,
1762  lazy_local_call: Option<u32>,
1763
1764  index: u32,
1765  arg1: u64,
1766  arg2: u64,
1767  arg3: u64,
1768  arg4: u64,
1769  arg5: u64,
1770}
1771
1772unsafe extern "C" fn tls_dispatcher(
1773  arg1: u64,
1774  arg2: u64,
1775  arg3: u64,
1776  arg4: u64,
1777  arg5: u64,
1778  index: std::os::raw::c_uint,
1779  _cookie: *mut std::os::raw::c_void,
1780) -> u64 {
1781  let yielder = ACTIVE_JIT_CODE_ZONE
1782    .with(|x| x.yielder.get())
1783    .expect("no yielder");
1784  let yielder = yielder.as_ref();
1785  let ret = yielder.suspend(Dispatch {
1786    async_preemption: false,
1787    memory_access_error: None,
1788    lazy_local_call: None,
1789    index,
1790    arg1,
1791    arg2,
1792    arg3,
1793    arg4,
1794    arg5,
1795  });
1796  ret
1797}
1798
1799unsafe extern "C" fn tls_local_call_resolver(resolver_id: std::os::raw::c_uint) -> u64 {
1800  let program = ACTIVE_PROGRAM.with(|x| x.get());
1801  if !program.is_null() {
1802    if let Some(ptr) = (*program).cached_resolver_target(resolver_id) {
1803      return ptr as u64;
1804    }
1805  }
1806
1807  let yielder = ACTIVE_JIT_CODE_ZONE
1808    .with(|x| x.yielder.get())
1809    .expect("no yielder");
1810  let yielder = yielder.as_ref();
1811  yielder.suspend(Dispatch {
1812    lazy_local_call: Some(resolver_id),
1813    ..Default::default()
1814  })
1815}
1816
1817unsafe extern "C" fn std_validator(
1818  index: std::os::raw::c_uint,
1819  _vm: *const crate::ubpf::ubpf_vm,
1820) -> bool {
1821  let loader = LOADING_PROGRAM_LOADER.with(|x| x.get());
1822  if loader.is_null() {
1823    return false;
1824  }
1825  let loader = &*loader;
1826  let entropy = (index >> 16) & 0xffff;
1827  let index = (((index & 0xffff) as u16) ^ loader.helper_id_xor).wrapping_sub(1);
1828  loader.helpers.get(index as usize).map(|x| x.0) == Some(entropy as u16)
1829}
1830
1831#[cfg(all(target_arch = "x86_64", target_os = "linux"))]
1832unsafe fn program_counter(uctx: *mut libc::ucontext_t) -> usize {
1833  (*uctx).uc_mcontext.gregs[libc::REG_RIP as usize] as usize
1834}
1835
1836#[cfg(all(target_arch = "aarch64", target_os = "linux"))]
1837unsafe fn program_counter(uctx: *mut libc::ucontext_t) -> usize {
1838  (*uctx).uc_mcontext.pc as usize
1839}
1840
1841unsafe extern "C" fn sigsegv_handler(
1842  _sig: i32,
1843  siginfo: *mut libc::siginfo_t,
1844  uctx: *mut libc::ucontext_t,
1845) {
1846  let fail = || restore_default_signal_handler(libc::SIGSEGV);
1847
1848  let Some((jit_code_zone, pointer_cage, yielder)) = ACTIVE_JIT_CODE_ZONE.with(|x| {
1849    if x.valid.load(Ordering::Relaxed) {
1850      compiler_fence(Ordering::Acquire);
1851      Some((
1852        x.code_range.get(),
1853        x.pointer_cage_protected_range.get(),
1854        x.yielder.get(),
1855      ))
1856    } else {
1857      None
1858    }
1859  }) else {
1860    return fail();
1861  };
1862
1863  let pc = program_counter(uctx);
1864
1865  if pc < jit_code_zone.0 || pc >= jit_code_zone.1 {
1866    return fail();
1867  }
1868
1869  // SEGV_MAPERR or SEGV_ACCERR.
1870  if (*siginfo).si_code != 1 && (*siginfo).si_code != 2 {
1871    return fail();
1872  }
1873
1874  let si_addr = (*siginfo).si_addr() as usize;
1875  if si_addr < pointer_cage.0 || si_addr >= pointer_cage.1 {
1876    return fail();
1877  }
1878
1879  let yielder = yielder.expect("no yielder").as_ref();
1880  yielder.suspend(Dispatch {
1881    memory_access_error: Some(si_addr),
1882    ..Default::default()
1883  });
1884}
1885
1886unsafe extern "C" fn sigusr1_handler(
1887  _sig: i32,
1888  _siginfo: *mut libc::siginfo_t,
1889  uctx: *mut libc::ucontext_t,
1890) {
1891  SIGUSR1_COUNTER.with(|x| x.set(x.get() + 1));
1892
1893  let Some((jit_code_zone, yielder)) = ACTIVE_JIT_CODE_ZONE.with(|x| {
1894    if x.valid.load(Ordering::Relaxed) {
1895      compiler_fence(Ordering::Acquire);
1896      Some((x.code_range.get(), x.yielder.get()))
1897    } else {
1898      None
1899    }
1900  }) else {
1901    return;
1902  };
1903  let pc = program_counter(uctx);
1904  if pc < jit_code_zone.0 || pc >= jit_code_zone.1 {
1905    return;
1906  }
1907
1908  let yielder = yielder.expect("no yielder").as_ref();
1909  yielder.suspend(Dispatch {
1910    async_preemption: true,
1911    ..Default::default()
1912  });
1913}
1914
1915unsafe fn restore_default_signal_handler(signum: i32) {
1916  let act = libc::sigaction {
1917    sa_sigaction: libc::SIG_DFL,
1918    sa_flags: libc::SA_SIGINFO,
1919    sa_mask: std::mem::zeroed(),
1920    sa_restorer: None,
1921  };
1922  if libc::sigaction(signum, &act, std::ptr::null_mut()) != 0 {
1923    libc::abort();
1924  }
1925}
1926
1927fn get_blocked_sigset() -> libc::sigset_t {
1928  unsafe {
1929    let mut s: libc::sigset_t = std::mem::zeroed();
1930    libc::sigaddset(&mut s, libc::SIGUSR1);
1931    libc::sigaddset(&mut s, libc::SIGSEGV);
1932    s
1933  }
1934}