```{index} single: Peta::FFI::Queue ``` # Peta::FFI::Queue ```{pperl-module-badges} Peta::FFI::Queue ``` 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](../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](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 ```perl 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.