File Coverage

blib/lib/Hypersonic/Event/IOCP.pm

Criterion	Covered	Total	%
statement	12	32	37.5
branch			n/a
condition			n/a
subroutine	5	18	27.7
pod	13	14	92.8
total	30	64	46.8

package Hypersonic::Event::IOCP;

use strict;
use warnings;
use 5.010;

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

our $VERSION = '0.15';

sub name { 'iocp' }

sub available {
    return $^O eq 'MSWin32';
}

sub includes {
    return <<'C';
#include
#include
#include
#pragma comment(lib, "ws2_32.lib")
#pragma comment(lib, "mswsock.lib")
C
}

sub defines {
    return <<'C';
#define EV_BACKEND_IOCP 1
#ifndef MAX_EVENTS
#define MAX_EVENTS 1024
#endif

/* Operation types for OVERLAPPED tracking */
#define OP_ACCEPT  1
#define OP_READ    2
#define OP_WRITE   3

/* Per-I/O data structure */
typedef struct {
    OVERLAPPED overlapped;
    WSABUF wsa_buf;
    char buffer[65536];
    int op_type;
    SOCKET socket;
} PER_IO_DATA;

/* AcceptEx function pointer (loaded dynamically) */
static LPFN_ACCEPTEX lpfnAcceptEx = NULL;
C
}

sub event_struct { 'OVERLAPPED_ENTRY' }

sub extra_cflags  { '' }
sub extra_ldflags { '-lws2_32 -lmswsock' }

# IOCP is completion-based, fundamentally different from readiness-based APIs
sub gen_create {
    my ($class, $builder, $listen_fd_var) = @_;

    $builder->comment('IOCP backend - Windows I/O Completion Ports')
      ->comment('High-performance completion-based I/O')
      ->blank
      ->comment('Initialize Winsock')
      ->line('WSADATA wsa_data;')
      ->if('WSAStartup(MAKEWORD(2,2), &wsa_data) != 0')
        ->line('croak("WSAStartup failed");')
      ->endif
      ->blank
      ->comment('Create I/O Completion Port')
      ->line('HANDLE iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);')
      ->if('iocp == NULL')
        ->line('WSACleanup();')
        ->line('croak("CreateIoCompletionPort failed");')
      ->endif
      ->blank
      ->comment('Associate listen socket with IOCP')
      ->line("if (CreateIoCompletionPort((HANDLE)$listen_fd_var, iocp, (ULONG_PTR)$listen_fd_var, 0) == NULL) {")
      ->line('    CloseHandle(iocp);')
      ->line('    WSACleanup();')
      ->line('    croak("Failed to associate listen socket with IOCP");')
      ->line('}')
      ->blank
      ->comment('Load AcceptEx function')
      ->line('GUID guid_acceptex = WSAID_ACCEPTEX;')
      ->line('DWORD bytes;')
      ->line("if (WSAIoctl($listen_fd_var, SIO_GET_EXTENSION_FUNCTION_POINTER,")
      ->line('             &guid_acceptex, sizeof(guid_acceptex),')
      ->line('             &lpfnAcceptEx, sizeof(lpfnAcceptEx),')
      ->line('             &bytes, NULL, NULL) == SOCKET_ERROR) {')
      ->line('    CloseHandle(iocp);')
      ->line('    WSACleanup();')
      ->line('    croak("Failed to load AcceptEx");')
      ->line('}')
      ->blank
      ->comment('Post initial AcceptEx')
      ->line('PER_IO_DATA* accept_data = (PER_IO_DATA*)calloc(1, sizeof(PER_IO_DATA));')
      ->line('accept_data->op_type = OP_ACCEPT;')
      ->line('accept_data->socket = WSASocket(AF_INET, SOCK_STREAM, 0, NULL, 0, WSA_FLAG_OVERLAPPED);')
      ->line("lpfnAcceptEx($listen_fd_var, accept_data->socket, accept_data->buffer,")
      ->line('             0, sizeof(struct sockaddr_in) + 16, sizeof(struct sockaddr_in) + 16,')
      ->line('             &bytes, &accept_data->overlapped);')
      ->blank
      ->line('int ev_fd = (int)(intptr_t)iocp;')  # Store IOCP handle as ev_fd
      ->blank;
}

# Associate socket with IOCP and post read operation
sub gen_add {
    my ($class, $builder, $loop_var, $fd_var) = @_;

    $builder->line("HANDLE iocp_handle = (HANDLE)(intptr_t)$loop_var;")
      ->line("CreateIoCompletionPort((HANDLE)$fd_var, iocp_handle, (ULONG_PTR)$fd_var, 0);")
      ->blank
      ->comment('Post initial read operation')
      ->line('PER_IO_DATA* io_data = (PER_IO_DATA*)calloc(1, sizeof(PER_IO_DATA));')
      ->line('io_data->op_type = OP_READ;')
      ->line("io_data->socket = $fd_var;")
      ->line('io_data->wsa_buf.buf = io_data->buffer;')
      ->line('io_data->wsa_buf.len = sizeof(io_data->buffer);')
      ->line('DWORD flags = 0;')
      ->line("WSARecv($fd_var, &io_data->wsa_buf, 1, NULL, &flags, &io_data->overlapped, NULL);");
}

# Cancel pending I/O and close socket
sub gen_del {
    my ($class, $builder, $loop_var, $fd_var) = @_;

    $builder->line("CancelIo((HANDLE)$fd_var);")
      ->line("closesocket($fd_var);");
}

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

    $builder->line("HANDLE iocp_handle = (HANDLE)(intptr_t)$loop_var;")
      ->line("static OVERLAPPED_ENTRY $events_var" . "[MAX_EVENTS];")
      ->line("ULONG $count_var = 0;")
      ->blank
      ->line("BOOL ok = GetQueuedCompletionStatusEx(iocp_handle, $events_var, MAX_EVENTS, &$count_var, $timeout_var, FALSE);")
      ->if('!ok')
        ->if('GetLastError() == WAIT_TIMEOUT')
          ->line('continue;')
        ->endif
        ->line('perror("GetQueuedCompletionStatusEx");')
        ->line('break;')
      ->endif;
}

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

    $builder->line("OVERLAPPED_ENTRY* entry = &${events_var}[$index_var];")
      ->line('PER_IO_DATA* io_data = CONTAINING_RECORD(entry->lpOverlapped, PER_IO_DATA, overlapped);')
      ->line("int $fd_var = (int)io_data->socket;")
      ->line('int op_type = io_data->op_type;')
      ->line('DWORD bytes_transferred = entry->dwNumberOfBytesTransferred;')
      ->blank
      ->if('op_type == OP_ACCEPT')
        ->comment('Accept completed - io_data->socket is the new client')
        ->line("$fd_var = listen_fd;")  # Signal this was accept
        ->line('int client_fd = (int)io_data->socket;')
      ->elsif('op_type == OP_READ')
        ->if('bytes_transferred == 0')
          ->comment('Connection closed')
          ->line("closesocket($fd_var);")
          ->line('free(io_data);')
          ->line('g_active_connections--;')
          ->line('continue;')
        ->endif
        ->comment('Data received - already in io_data->buffer')
      ->endif;
}

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

    $builder->line('CloseHandle(iocp);')
      ->line('WSACleanup();');
}

# Future/Pool integration - IOCP uses different mechanism for pool notify
# On Windows, pool uses a pipe which can be added to IOCP via overlapped read
sub gen_add_pool_notify {
    my ($class, $builder, $loop_var, $notify_fd_var) = @_;

    $builder->line("/* Add pool notify fd to IOCP */")
            ->line("HANDLE iocp_handle = (HANDLE)(intptr_t)$loop_var;")
            ->line("CreateIoCompletionPort((HANDLE)$notify_fd_var, iocp_handle, (ULONG_PTR)$notify_fd_var, 0);")
            ->line('PER_IO_DATA* pool_io = (PER_IO_DATA*)calloc(1, sizeof(PER_IO_DATA));')
            ->line('pool_io->op_type = 4;')  # New type for pool notify
            ->line("pool_io->socket = $notify_fd_var;")
            ->line('pool_io->wsa_buf.buf = pool_io->buffer;')
            ->line('pool_io->wsa_buf.len = 1;')
            ->line('DWORD flags = 0;')
            ->line("ReadFile((HANDLE)$notify_fd_var, pool_io->buffer, 1, NULL, &pool_io->overlapped);");
}

1;

__END__