File Coverage

blib/lib/Hypersonic/Event/IOUring.pm

Criterion	Covered	Total	%
statement	18	54	33.3
branch	2	16	12.5
condition	4	14	28.5
subroutine	6	23	26.0
pod	13	19	68.4
total	43	126	34.1

package Hypersonic::Event::IOUring;

use strict;
use warnings;
use 5.010;

use parent 'Hypersonic::Event::Role';

our $VERSION = '0.15';

sub name { 'io_uring' }

sub available {
    return 0 unless $^O eq 'linux';

    # Check kernel version >= 5.1
    my $ver = `uname -r 2>/dev/null` || '';
    my ($major, $minor) = $ver =~ /^(\d+)\.(\d+)/;
    return 0 unless $major && ($major > 5 || ($major == 5 && $minor >= 1));

    # Check for liburing headers
    my $has_header = -f '/usr/include/liburing.h'
        || -f '/usr/local/include/liburing.h'
        || -f '/usr/include/x86_64-linux-gnu/liburing.h';
    return 0 unless $has_header;

    # io_uring may be disabled at the kernel level. RHEL9 ships with
    # kernel.io_uring_disabled=2 by default; a value of 1 or 2 means
    # the syscall returns EINVAL/EPERM regardless of liburing being
    # linkable. Bail before we sink time into a compile+link probe.
    if (open my $fh, '<', '/proc/sys/kernel/io_uring_disabled') {
        my $disabled = <$fh>;
        close $fh;
        chomp $disabled if defined $disabled;
        return 0 if defined $disabled && $disabled ne '0';
    }

    # Compile-link-and-RUN probe. A pure link check passes on systems
    # that have liburing installed but where io_uring_setup() will
    # nevertheless fail at runtime (kernel disabled, sandboxing, missing
    # liburing.so at exec time). Actually open and close a ring.
    require Hypersonic::JIT::Util;
    return Hypersonic::JIT::Util->can_run(
        '',
        '-luring',
        'struct io_uring ring; int rc = io_uring_queue_init(8, &ring, 0); '
            . 'if (rc < 0) return 1; io_uring_queue_exit(&ring); return 0;',
        '#include ',
    );
}

sub includes {
    # liburing.h for the server loop.  is needed for the
    # UA::Async slot-tracking helpers (gen_create_loop / _add_with_slot /
    # _get_slot) - io_uring is Linux 5.1+ which always has epoll.
    return "#include \n#include ";
}

sub defines {
    return <<'C';
#define EV_BACKEND_IO_URING 1
#ifndef URING_ENTRIES
#define URING_ENTRIES 256
#endif
#ifndef MAX_EVENTS
#define MAX_EVENTS 1024
#endif

/* User data encoding: type in high bits, fd in low bits */
#define UD_ACCEPT 0x10000000
#define UD_READ   0x20000000
#define UD_WRITE  0x30000000
#define UD_FD_MASK 0x0FFFFFFF
C
}

sub event_struct { 'io_uring_cqe' }

# UA::Async slot-tracking helpers below use a private epoll instance
# (io_uring's own slot tracking would mean weaving user_data through
# the shared submission ring, which is a lot more invasive). So when
# UA::Async asks for a buffer to pass to gen_wait_once it must be
# struct epoll_event[], NOT io_uring_cqe[]. See the Fedora 43 / perl
# 5.38.5 smoker report (5ce1e632) for what happens when the wrong
# type is declared - epoll_wait() argument-type mismatch + a missing
# `data` member access.
sub slot_event_struct { 'epoll_event' }

sub extra_cflags  { '' }
sub extra_ldflags { '-luring' }

# io_uring is fundamentally different - uses submission/completion queues
sub gen_create {
    my ($class, $builder, $listen_fd_var) = @_;

    $builder->comment('io_uring backend - high performance Linux I/O')
      ->line('static struct io_uring ring;')
      ->line('static int ring_initialized = 0;')
      ->blank
      ->if('!ring_initialized')
        ->if('io_uring_queue_init(URING_ENTRIES, &ring, 0) < 0')
          # gen_create is inlined into hypersonic_run_event_loop, which
          # is a void XS function. `return ;` triggers GCC 14+
          # -Wreturn-mismatch (now an error). croak from XS instead -
          # it longjmps out cleanly and surfaces a Perl-level error.
          ->line('croak("io_uring_queue_init() failed: %s", strerror(errno));')
        ->endif
        ->line('ring_initialized = 1;')
      ->endif
      ->blank
      ->comment('Submit initial accept')
      ->line('struct io_uring_sqe* sqe = io_uring_get_sqe(&ring);')
      ->if('sqe')
        ->line("io_uring_prep_accept(sqe, $listen_fd_var, NULL, NULL, 0);")
        ->line("io_uring_sqe_set_data(sqe, (void*)(uintptr_t)(UD_ACCEPT | $listen_fd_var));")
        ->line('io_uring_submit(&ring);')
      ->endif
      ->line('int ev_fd = 0;')  # Dummy - io_uring uses ring structure
      ->blank;
}

# Submit read operation
sub gen_add {
    my ($class, $builder, $loop_var, $fd_var) = @_;

    $builder->line('sqe = io_uring_get_sqe(&ring);')
      ->if('sqe')
        ->line("io_uring_prep_recv(sqe, $fd_var, recv_buf, RECV_BUF_SIZE, 0);")
        ->line("io_uring_sqe_set_data(sqe, (void*)(uintptr_t)(UD_READ | $fd_var));")
        ->line('io_uring_submit(&ring);')
      ->endif;
}

# Cancel pending operations
sub gen_del {
    my ($class, $builder, $loop_var, $fd_var) = @_;

    $builder->comment('io_uring: close fd (pending ops will complete with error)')
      ->line("close($fd_var);");
}

# Wait for completions
sub gen_wait {
    my ($class, $builder, $loop_var, $events_var, $count_var, $timeout_var) = @_;

    $builder->line('struct io_uring_cqe* cqe;')
      ->line('struct __kernel_timespec ts;')
      ->line("ts.tv_sec = $timeout_var / 1000;")
      ->line("ts.tv_nsec = ($timeout_var % 1000) * 1000000;")
      ->blank
      ->comment('Wait for at least one completion')
      ->line('int wait_result = io_uring_wait_cqe_timeout(&ring, &cqe, &ts);')
      ->if('wait_result < 0')
        ->if('wait_result == -ETIME')
          ->line('continue;')  # Timeout is normal
        ->endif
        ->if('wait_result == -EINTR')
          ->line('continue;')
        ->endif
        ->line('break;')
      ->endif
      ->blank
      ->comment('Process all available completions')
      ->line('unsigned head;')
      ->line("int $count_var = 0;")
      ->line("static struct io_uring_cqe* cqes[MAX_EVENTS];")
      ->line('io_uring_for_each_cqe(&ring, head, cqe) {')
      ->line("    if ($count_var < MAX_EVENTS) cqes[$count_var++] = cqe;")
      ->line('}')
      ->line("$events_var = cqes;");  # Point to our array
}

# Extract operation type and fd from completion
sub gen_get_fd {
    my ($class, $builder, $events_var, $index_var, $fd_var) = @_;

    $builder->line("struct io_uring_cqe* completion = ${events_var}[$index_var];")
      ->line('uintptr_t user_data = (uintptr_t)io_uring_cqe_get_data(completion);')
      ->line('int op_type = user_data & 0xF0000000;')
      ->line("int $fd_var = user_data & UD_FD_MASK;")
      ->line('int result = completion->res;')
      ->blank
      ->comment('Mark completion as seen')
      ->line('io_uring_cqe_seen(&ring, completion);')
      ->blank
      ->comment('Handle based on operation type')
      ->if('op_type == UD_ACCEPT')
        ->if('result >= 0')
          ->comment('result is the new client fd')
          ->line("int client_fd = result;")
          ->line("$fd_var = listen_fd;")  # Signal this was accept
        ->else
          ->line('continue;')  # Accept failed
        ->endif
      ->elsif('op_type == UD_READ')
        ->if('result <= 0')
          ->comment('Connection closed or error')
          ->line("close($fd_var);")
          ->line('g_active_connections--;')
          ->line('continue;')
        ->endif
        ->comment('result is bytes read - already in recv_buf')
      ->endif;
}

# Cleanup io_uring resources
sub gen_cleanup {
    my ($class, $builder) = @_;

    $builder->if('ring_initialized')
      ->line('io_uring_queue_exit(&ring);')
      ->line('ring_initialized = 0;')
    ->endif;
}

# ============================================================
# Async Slot Integration Methods (UA Async)
#
# UA::Async tracks per-slot fd readiness independently of the server
# loop's io_uring ring. We use epoll under the hood since io_uring
# requires Linux 5.1+ which always has epoll, and a separate epoll
# instance is much simpler than weaving slot tracking through the
# server's submission ring.
# ============================================================

sub gen_wait_once {
    my ($class, $builder, $loop_var, $events_var, $count_var, $timeout_ms) = @_;

    $builder->line("$count_var = epoll_wait($loop_var, $events_var, MAX_EVENTS, $timeout_ms);")
      ->line("if ($count_var < 0 && errno == EINTR) $count_var = 0;");
}

sub gen_create_loop {
    my ($class, $builder, $loop_var) = @_;

    $builder->line("$loop_var = epoll_create1(0);")
      ->if("$loop_var < 0")
        ->line('croak("epoll_create1() failed");')
      ->endif;
}

sub gen_add_with_slot {
    my ($class, $builder, $loop_var, $fd_var, $slot_var, $events) = @_;

    my $ev_flags = $events eq 'read' ? 'EPOLLIN | EPOLLET | EPOLLONESHOT'
                 : $events eq 'write' ? 'EPOLLOUT | EPOLLET | EPOLLONESHOT'
                 : 'EPOLLIN | EPOLLET | EPOLLONESHOT';

    $builder->line('{')
      ->line('    struct epoll_event _ev;')
      ->line("    _ev.events = $ev_flags;")
      ->line("    _ev.data.u32 = (uint32_t)$slot_var;")
      ->line("    epoll_ctl($loop_var, EPOLL_CTL_ADD, $fd_var, &_ev);")
      ->line('}');
}

sub gen_get_slot {
    my ($class, $builder, $events_var, $index_var, $slot_var) = @_;

    $builder->line("int $slot_var;")
      ->line("$slot_var = (int)${events_var}[$index_var].data.u32;");
}

# When async-slot helpers above are emitted, the io_uring includes
# already pull in  via the generated Hypersonic includes.
# We don't need to add it here.

# Future/Pool integration - add pool notify fd via poll on the fd
sub gen_add_pool_notify {
    my ($class, $builder, $loop_var, $notify_fd_var) = @_;

    $builder->line("/* Add pool notify fd to io_uring via poll */")
            ->line('sqe = io_uring_get_sqe(&ring);')
            ->if('sqe')
              ->line("io_uring_prep_poll_add(sqe, $notify_fd_var, POLLIN);")
              ->line("io_uring_sqe_set_data(sqe, (void*)(uintptr_t)(0x40000000 | $notify_fd_var));")
              ->line('io_uring_submit(&ring);')
            ->endif;
}

1;

__END__