view release on metacpan or  search on metacpan

lib/Async/Event/Interval.pm  view on Meta::CPAN

sub _events_write {
    my ($cb) = @_;

    my $knot = tied(%events);

    return $cb->() unless $knot;

    my $callback_result;

    $knot->lock(LOCK_EX, sub {
        $callback_result = $cb->();
    });

    return $callback_result;
}
sub _pid {
    my ($self, $pid) = @_;
    if (defined $pid) {
        $self->{pid} = $pid;
        _events_write(sub { $events{$self->id}->{pid} = $self->{pid} });
    }
    return $self->{pid} || undef;
}
sub _pm {
    my ($self) = @_;

    if (! exists $self->{pm}) {
        $self->{pm} = Parallel::ForkManager->new(1);
    }

    return $self->{pm};
}
sub _rand_shm_key {
    return sprintf('0x%x', int(rand(0x7FFFFFFF)));
}
sub _rand_shm_lock {
    # Used for the 'protected' option in the %events hash creation.
    #
    # IPC::Shareable 1.14+ persists 'protected' in a semaphore slot
    # (SEM_PROTECTED), which the system caps at semvmx (typically 0..32767, and
    # 0 means "unprotected"). Derive a stable, in-range value from $$ so a
    # forked subprocess inherits the same key.

    return 1 + ($$ % 32767);
}
sub _run_callback {
    my ($self, @params) = @_;

    my $timeout = $self->timeout;

    my $ok = eval {
        if ($timeout) {
            my $handler = sub {
                die "Callback timed out after ${timeout} seconds\n"
            };

            local $SIG{ALRM} = $handler;

            # Re-install SIGALRM via POSIX::sigaction with flags=0 to
            # explicitly clear SA_RESTART. Perl's default $SIG{ALRM} setup
            # leaves SA_RESTART on, which causes the kernel to transparently
            # resume select() and other restartable syscalls after SIGALRM —
            # silently swallowing the timeout on Linux (and anywhere SA_RESTART
            # is the default). The local $SIG{ALRM} above still does the
            # safe-signal dispatch to the Perl coderef; sigaction just fixes the
            # kernel flags.

            my $sigset = POSIX::SigSet->new(POSIX::SIGALRM());
            my $sa     = POSIX::SigAction->new($handler, $sigset, 0);
            my $old    = POSIX::SigAction->new();
            POSIX::sigaction(POSIX::SIGALRM(), $sa, $old);

            alarm($timeout);
            $self->_cb->(@params);
            alarm(0);

            POSIX::sigaction(POSIX::SIGALRM(), $old);
        }
        else {
            $self->_cb->(@params);
        }
        1;
    };

    alarm(0) if $timeout;

    if (! $ok) {
        my $err = $@;

        $self->_errors(1);
        $self->_error_message($err);
        $self->_runs(1);
        $self->status;

        die $err;
    }

    $self->_runs(1);
    $self->status;
}
sub _runs {
    my ($self, $increment) = @_;
    if (defined $increment) {
        _events_write(sub { $events{$self->id}->{runs}++ });
    }
    return _events_read(sub { $events{$self->id}->{runs} });
}
sub _setup {
    my ($self, $interval, $cb, @args) = @_;
    $self->interval($interval);
    $self->_cb($cb);
    $self->_args(\@args);
}
sub _shm_lock {
    return $shared_memory_protect_lock;
}
sub _signal_and_wait {
    my ($self, $sig, $timeout) = @_;

    kill $sig, $self->pid;

lib/Async/Event/Interval.pm  view on Meta::CPAN

Optional, Integer: The number of whole seconds the callback is allowed to
execute for before timing out. Must be a non-negative integer; fractional
seconds are not supported. Use C<0> or C<undef> to disable.

Default: C<0>

Return: Currently set value.

B<Note>: The timeout is read from shared memory at the start of every callback
invocation, so changes made via this setter while an event is running take
effect on the next iteration of the interval loop (mirroring
L<interval()|/interval($seconds)>).

=head2 immediate($value)

Sets (or gets) whether the callback fires immediately on
L<start()|/start(@params)>, bypassing the first interval wait. Subsequent
invocations follow the normal interval cadence.

Parameters:

    $value

Optional, Integer: C<1> to enable immediate first execution, C<0> or C<undef>
to disable. Must be a non-negative integer when defined.

Default: C<0>

Return: Currently set value.

B<Note>: The flag is read from shared memory on each iteration of the event
loop. Changes made before calling the initial L<start()|/start(@params)> always
take effect. Changes made after C<start()> take effect on the next loop
iteration; however, once the first callback has executed, C<immediate> has
already served its purpose and further changes will not trigger another
immediate execution. Restart the event for a fresh C<immediate> check.

B<Note>: This feature is a no-op when running in single run mode. In that mode,
the event is always fired immediately on a call to C<start()>.

=head2 shared_scalar

Returns a reference to a scalar variable that can be shared between the main
process and the events. This reference can be used within multiple events, and
multiple shared scalars can be created by each event.

To read from or assign to the returned scalar, dereference it:

    $$s = 42;              # plain number
    $$s = 'some string';   # plain string
    $$s = { key => 'v' };  # hashref
    $$s = [1, 2, 3];       # arrayref

B<Supported values>: Internally L<IPC::Shareable> serializes to JSON by
default, so values must be JSON-representable: scalars (strings/numbers),
arrayrefs, hashrefs, and combinations of those. Blessed objects, code
references, regex references, and globs are B<not> supported and will be
silently lost or corrupt the segment.

Nested references work transparently and cleanup is automatic. Note that
under the hood, each nested hashref/arrayref allocates its own child
shared-memory segment, so very deeply nested structures consume one shm
segment per node:

    $$s = { config => { db => { host => 'localhost', port => 5432 } } };

    my $host = $$s->{config}{db}{host};   # 'localhost'

B<Updating a stored hashref>: When extending a hashref already in the scalar,
mutate through the dereference directly. Do not fetch the reference into a
lexical, mutate it, and store it back: that pattern corrupts the segment
because the fetched reference still carries C<IPC::Shareable>'s tied magic,
and re-storing a tied value into its own parent breaks the serialization:

    # Recommended: direct dereferenced mutation
    $$s->{new_key} = 'val';

    # Also works (modern stacks): spread + reassign
    $$s = { %{$$s}, new_key => 'val' };

    # Unreliable: re-storing a fetched reference corrupts the segment
    # my $h = $$s; $h->{new_key} = 'val'; $$s = $h;

The spread idiom replaces the entire stored value, which on older
C<IPC::Shareable> versions can lose accumulated cross-process writes
(later writers replacing earlier writers' data). The direct dereferenced
mutation avoids the nested-segment STORE path and is the more portable
choice.

B<Lifetime>: The underlying shared memory segment is owned by the event object
that created it. When the event goes out of scope (and its C<DESTROY> runs),
every C<shared_scalar> it created is released. Do not dereference the returned
scalar reference after the owning event has been destroyed; the segment will no
longer exist. If you need a shared scalar whose lifetime is independent of any
event, tie it directly with L<IPC::Shareable>.

B<Hex keys>: L</info> and L</events> return C<shared_scalars> as an arrayref of
hex key strings. These identify the underlying IPC segments and can be used to
re-attach from another process:

    my $info = $event->info;

    for my $key (@{ $info->{shared_scalars} }) {
        tie my $scalar, 'IPC::Shareable', $key, {};
        print "$$scalar\n";
    }

In practice, however, it is simpler to retain the reference returned by
C<shared_scalar()> and use it directly.

=head1 METHODS - EVENT INFORMATION

=head2 errors

Returns the number of times a started or restarted event has crashed
unexpectedly. See L</error> to test whether the event is currently in an
error state.

=head2 error_message