Peta::FFI::Queue

📦 std

Callbacks a C library makes from threads of its own, queued for Perl’s event loop.

Some libraries call back on a thread of theirs: librados calls its monitor log callback and its watch and notify callbacks that way. Perl cannot run there, so a Peta::FFI closure called from such a thread stops the program. A queued closure runs no Perl when it is called: it copies its arguments, as its signature reads them, into a queue and returns at once, and Perl runs its sub later, when the event loop takes the queue’s events.

Peta::FFI::Queue->new makes a queue, and Peta::FFI::closure($sig, $sub, queue => $queue) a closure that queues into it. Every call of the closure, from any thread - Perl’s own included - becomes one event, { handler => $sub, args => [...], closure => $address }; dispatch runs each queued event’s handler with its arguments, oldest first. The queue reads ready on fd while it holds events, so it is a Peta::FFI::EventSource for any event loop.

A queued closure returns v, or a number, b or o given once as returns => $value (0 when not given), since nothing can compute a value at the time of the call. Its arguments are what any closure takes: numbers, b, o, p (copied up to its NUL) and records, copied, and bytes or an array of numbers C passes with their length in another argument, &P[#k] and &T[#k]. A P argument, an address valid only during the call, is not.

Synopsis

use Peta::FFI;
use Peta::FFI::Queue;
use Peta::FFI::EventSource;

my $queue = Peta::FFI::Queue->new;

# librados's rados_log_callback2_t
my $log = Peta::FFI::closure('(opppp u64u64u64 pp)v', sub {
    my (undef, $line, $channel, $who, $name, $sec, $nsec, $seq, $level, $msg) = @_;
    print "$level $who: $msg\n";
}, queue => $queue);
$monitor_log2->($cluster, "info", $log, undef);

# rados_watchcb2_t: the notify's payload, data_len bytes
my $watch = Peta::FFI::closure('(ou64u64u64&P[#6]L)v', sub {
    my (undef, $notify_id, $cookie, $notifier, $payload) = @_;
    ...
}, queue => $queue);

my $source = Peta::FFI::EventSource->new($queue,
    fd => \&Peta::FFI::Queue::fd, next => \&Peta::FFI::Queue::next);
my $guard = $source->watch(sub { $_->{handler}->(@{ $_->{args} }) for @_ });

# or, with a loop of your own
vec(my $rin = '', $queue->fd, 1) = 1;
$queue->dispatch while select(my $rout = $rin, undef, undef, undef);

Bounds and lifetimes

A queue holds at most limit events (Peta::FFI::Queue->new(limit => $n), 65536 when not given), so a library calling faster than Perl takes them cannot fill the memory: a call that finds the queue full is dropped, and lost counts it. An event whose closure was freed with closure_free before Perl took it is dropped too. A queued closure keeps its queue’s memory alive, so a library may call it after the queue object is gone; those calls are dropped. As for any closure, closure_free comes only after the library has been told to stop calling.

Functions

Queues

new

Peta::FFI::Queue->new(limit => $n): a queue of at most $n events, 65536 when not given.

dispatch

$queue->dispatch: run the handler of every queued call with its arguments, oldest first, and return how many ran. A handler that dies stops it there, with the error; the calls after it stay queued.

lost

$queue->lost: how many calls were dropped because the queue was full.

Events

fd

$queue->fd: a descriptor that reads ready while events are queued, for an event loop. Never close it: it is the queue’s.

next

$queue->next: the oldest queued call as a hash reference - its closure’s sub as handler, the arguments as the sub would have got them as args, the closure’s address as closure - or undef when there is none.