view release on metacpan or  search on metacpan

Fork.pm  view on Meta::CPAN

hanging around all the time to fork off some new processes, which might be
an advantage when there are long time spans where no extra processes are
needed.

Example:

   AnyEvent::Fork
      ->new_exec
      ->require ("Some::Module")
      ->run ("Some::Module::run", sub {
         my ($fork_fh) = @_;
      });

=back

=head1 THE C<AnyEvent::Fork> CLASS

This module exports nothing, and only implements a single class -
C<AnyEvent::Fork>.

There are two class constructors that both create new processes - C<new>
and C<new_exec>. The C<fork> method creates a new process by forking an
existing one and could be considered a third constructor.

Most of the remaining methods deal with preparing the new process, by
loading code, evaluating code and sending data to the new process. They
usually return the process object, so you can chain method calls.

If a process object is destroyed before calling its C<run> method, then
the process simply exits. After C<run> is called, all responsibility is
passed to the specified function.

As long as there is any outstanding work to be done, process objects
resist being destroyed, so there is no reason to store them unless you
need them later - configure and forget works just fine.

=over 4

=cut

package AnyEvent::Fork;

use common::sense;

use Errno ();

use AnyEvent;
use AnyEvent::Util ();

use IO::FDPass;

our $VERSION = 1.32;

# the early fork template process
our $EARLY;

# the empty template process
our $TEMPLATE;

sub QUEUE() { 0 }
sub FH()    { 1 }
sub WW()    { 2 }
sub PID()   { 3 }
sub CB()    { 4 }

sub _new {
   my ($self, $fh, $pid) = @_;

   AnyEvent::Util::fh_nonblocking $fh, 1;

   $self = bless [
      [],    # write queue - strings or fd's
      $fh,
      undef, # AE watcher
      $pid,
   ], $self;

   $self
}

sub _cmd {
   my $self = shift;

   # ideally, we would want to use "a (w/a)*" as format string, but perl
   # versions from at least 5.8.9 to 5.16.3 are all buggy and can't unpack
   # it.
   push @{ $self->[QUEUE] }, pack "a L/a*", $_[0], $_[1];

   $self->[WW] ||= AE::io $self->[FH], 1, sub {
      do {
         # send the next "thing" in the queue - either a reference to an fh,
         # or a plain string.

         if (ref $self->[QUEUE][0]) {
            # send fh
            unless (IO::FDPass::send fileno $self->[FH], fileno ${ $self->[QUEUE][0] }) {
               return if $! == Errno::EAGAIN || $! == Errno::EWOULDBLOCK;
               undef $self->[WW];
               die "AnyEvent::Fork: file descriptor send failure: $!";
            }

            shift @{ $self->[QUEUE] };

         } else {
            # send string
            my $len = syswrite $self->[FH], $self->[QUEUE][0];

            unless ($len) {
               return if $! == Errno::EAGAIN || $! == Errno::EWOULDBLOCK;
               undef $self->[WW];
               die "AnyEvent::Fork: command write failure: $!";
            }

            substr $self->[QUEUE][0], 0, $len, "";
            shift @{ $self->[QUEUE] } unless length $self->[QUEUE][0];
         }
      } while @{ $self->[QUEUE] };

      # everything written
      undef $self->[WW];

      # invoke run callback, if any
      if ($self->[CB]) {
         $self->[CB]->($self->[FH]);