view release on metacpan or  search on metacpan

lib/Async/Simple/Task/Fork.pm  view on Meta::CPAN

Possible keys for %all_optional_params:

    task         => coderef, function, called for each "data" passed to child process via $task->put( $data );

    timeout      => timeout in seconds between child checkings for new data passed. default 0.01

    kill_on_exit => kill (1) or not (0) subprocess on object destroy (1 by default).

=cut


=head2 BUILD

internal. Some tricks here:)

    1. Master process called $task->new with fork() inside
    2. After forking done we have two processes:
    2.1. Master gets one side of reader/writer pipes and pid of child
    2.2. Child - another side of pipes and extra logic with everlasting loop

=cut

sub BUILD {
    my ( $self ) = @_;

    # Return for master process
    # Only child tasks must go down and make a loop
    return $self  if $self->pid;

    # Child loop: untill parent is alive
    while ( 1 ) {
        $self->clear_answer;
        $self->get;

        unless ( $self->has_answer ) {
            sleep $self->timeout;
            next;
        }

        my $result = eval{ $self->task->( $self->answer ) };
        $self->clear_answer;
        $self->put( $result // $@ // '' );
    }

    exit(0);
};


=head2 fork_child

Makes child process and returns pid of child process to parent or 0 to child process

=cut

sub fork_child {
    my ( $self ) = @_;

    # This is here instead of BEGIN, because this package uses as "extends" in Async::Simple::Task::ForkTmpFile
    # TODO: Maybe it would be great to move this code(function) to separate package
    # if ( $^O =~ /^(dos|os2|MSWin32|NetWare)$/ ) {
    #     die 'Your OS does not support threads... Use Async::Simple::Task::ForkTmpFile instead.';
    # };

    # Pipes: parent -> child and child -> parent
    pipe my( $parent_reader, $child_writer  )  or die 'Child  to Parent pipe open error';
    pipe my( $child_reader,  $parent_writer )  or die 'Parent to Child  pipe open error';

    my $pid = fork() // die "fork() failed: $!";

    # child
    unless ( $pid ) {
        close $parent_reader;
        close $parent_writer;

        $child_writer->autoflush(1);

        $self->writer( $child_writer );
        $self->reader( $child_reader );

        # Important!
        # Just after that we trap into BUILD
        # with the infinitive loop for child process (pid=0)
        return 0;
    }

    # parent
    close $child_writer;
    close $child_reader;

    $parent_writer->autoflush(1);

    $self->writer( $parent_writer );
    $self->reader( $parent_reader );

    return $pid;
};


=head2 get

Reads from task, if something can be readed or returns undef after timeout.

    my $result = $task->get;

Please note! If your function can return an undef value, then you shoud check

    $task->has_result.

=cut

sub get {
    my ( $self ) = @_;

    my $pipe = $self->reader;
    my $data;

    # Try to read "marker" into data within timeout
    # Each pack starts with an empty line and serialized string of useful data.
    eval {
        local $SIG{ALRM} = sub { die "alarm\n" }; # NB: \n required
        alarm $self->timeout;