view release on metacpan or  search on metacpan

lib/AnyEvent/Util.pm  view on Meta::CPAN

            connect $r, $sa
               or next;

            accept my $w, $l
               or next;

            # vista has completely broken peername/sockname that return
            # fantasy ports. this combo seems to work, though.
            (Socket::unpack_sockaddr_in getpeername $r)[0]
            == (Socket::unpack_sockaddr_in getsockname $w)[0]
               or (($! = WSAEINVAL), next);

            # vista example (you can't make this shit up...):
            #(Socket::unpack_sockaddr_in getsockname $r)[0] == 53364
            #(Socket::unpack_sockaddr_in getpeername $r)[0] == 53363
            #(Socket::unpack_sockaddr_in getsockname $w)[0] == 53363
            #(Socket::unpack_sockaddr_in getpeername $w)[0] == 53365

            return ($r, $w);
         }

         ()
      };

      *portable_socketpair = \&_win32_socketpair;
      *portable_pipe       = \&_win32_socketpair;
   } else {
      *portable_pipe = sub () {
         my ($r, $w);

         pipe $r, $w
            or return;

         ($r, $w);
      };

      *portable_socketpair = sub () {
         socketpair my $fh1, my $fh2, Socket::AF_UNIX(), Socket::SOCK_STREAM(), 0
            or return;

         ($fh1, $fh2)
      };
   }
}

=item fork_call { CODE } @args, $cb->(@res)

Executes the given code block asynchronously, by forking. Everything the
block returns will be transferred to the calling process (by serialising and
deserialising via L<Storable>).

If there are any errors, then the C<$cb> will be called without any
arguments. In that case, either C<$@> contains the exception (and C<$!> is
irrelevant), or C<$!> contains an error number. In all other cases, C<$@>
will be C<undef>ined.

The code block must not ever call an event-polling function or use
event-based programming that might cause any callbacks registered in the
parent to run.

Win32 spoilers: Due to the endlessly sucky and broken native windows
perls (there is no way to cleanly exit a child process on that platform
that doesn't also kill the parent), you have to make sure that your main
program doesn't exit as long as any C<fork_calls> are still in progress,
otherwise the program won't exit. Also, on most windows platforms some
memory will leak for every invocation. We are open for improvements that
don't require XS hackery.

Note that forking can be expensive in large programs (RSS 200MB+). On
windows, it is abysmally slow, do not expect more than 5..20 forks/s on
that sucky platform (note this uses perl's pseudo-threads, so avoid those
like the plague).

Example: poor man's async disk I/O (better use L<AnyEvent::IO> together
with L<IO::AIO>).

   fork_call {
      open my $fh, "</etc/passwd"
         or die "passwd: $!";
      local $/;
      <$fh>
   } sub {
      my ($passwd) = @_;
      ...
   };

=item $AnyEvent::Util::MAX_FORKS [default: 10]

The maximum number of child processes that C<fork_call> will fork in
parallel. Any additional requests will be queued until a slot becomes free
again.

The environment variable C<PERL_ANYEVENT_MAX_FORKS> is used to initialise
this value.

=cut

our $MAX_FORKS = int 1 * $ENV{PERL_ANYEVENT_MAX_FORKS};
$MAX_FORKS = 10 if $MAX_FORKS <= 0;

my $forks;
my @fork_queue;

sub _fork_schedule;
sub _fork_schedule {
   require Storable unless $Storable::VERSION;
   require POSIX    unless $POSIX::VERSION;

   while ($forks < $MAX_FORKS) {
      my $job = shift @fork_queue
         or last;

      ++$forks;

      my $coderef = shift @$job;
      my $cb = pop @$job;
      
      # gimme a break...
      my ($r, $w) = portable_pipe
         or ($forks and last) # allow failures when we have at least one job
         or die "fork_call: $!";

      my $pid = fork;

      if ($pid != 0) {
         # parent
         close $w;

         my $buf;

         my $ww; $ww = AE::io $r, 0, sub {
            my $len = sysread $r, $buf, 65536, length $buf;

            return unless defined $len or $! != Errno::EINTR;

            if (!$len) {
               undef $ww;
               close $r;
               --$forks;
               _fork_schedule;
               
               my $result = eval { Storable::thaw ($buf) };
               $result = [$@] unless $result;
               $@ = shift @$result;

               $cb->(@$result);

               # work around the endlessly broken windows perls
               kill 9, $pid if AnyEvent::WIN32;

               # clean up the pid
               waitpid $pid, 0;
            }
         };

      } elsif (defined $pid) {
         # child
         close $r;

         my $result = eval {
            local $SIG{__DIE__};

            Storable::freeze ([undef, $coderef->(@$job)])
         };

         $result = Storable::freeze (["$@"])
            if $@;

         # windows forces us to these contortions
         my $ofs;

         while () {
            my $len = (length $result) - $ofs
               or last;

            $len = syswrite $w, $result, $len < 65536 ? $len : 65536, $ofs;

            last unless $len || (!defined $len && $! == Errno::EINTR);

            $ofs += $len;
         }

         # on native windows, _exit KILLS YOUR FORKED CHILDREN!
         if (AnyEvent::WIN32) {
            shutdown $w, 1; # signal parent to please kill us
            sleep 10; # give parent a chance to clean up
            sysread $w, (my $buf), 1; # this *might* detect the parent exiting in some cases.
         }
         POSIX::_exit (0);
         exit 1;
         
      } elsif (($! != &Errno::EAGAIN && $! != &Errno::EWOULDBLOCK && $! != &Errno::ENOMEM) || !$forks) {
         # we ignore some errors as long as we can run at least one job
         # maybe we should wait a few seconds and retry instead
         die "fork_call: $!";
      }
   }
}

sub fork_call(&@) {
   push @fork_queue, [@_];
   _fork_schedule;
}

END {
   if (AnyEvent::WIN32) {
      while ($forks) {
         @fork_queue = ();

lib/AnyEvent/Util.pm  view on Meta::CPAN

=item "fd>" => $see_above

Like ">", but redirects the specified fd number instead.

=item "<" => $see_above

The same, but redirects the program's standard input instead. The same
forms as for ">" are allowed.

In the callback form, the callback is supposed to return data to be
written, or the empty list or C<undef> or a zero-length scalar to signal
EOF.

Similarly, either the write data must be exhausted or an error is to be
signalled before the condvar is signalled, for both string-reference and
callback forms.

=item "fd<" => $see_above

Like "<", but redirects the specified file descriptor instead.

=item on_prepare => $cb

Specify a callback that is executed just before the command is C<exec>'ed,
in the child process. Be careful not to use any event handling or other
services not available in the child.

This can be useful to set up the environment in special ways, such as
changing the priority of the command or manipulating signal handlers (e.g.
setting C<SIGINT> to C<IGNORE>).

=item close_all => $boolean

When C<close_all> is enabled (default is disabled), then all extra file
descriptors will be closed, except the ones that were redirected and C<0>,
C<1> and C<2>.

See C<close_all_fds_except> for more details.

=item '$$' => \$pid

A reference to a scalar which will receive the PID of the newly-created
subprocess after C<run_cmd> returns.

Note the the PID might already have been recycled and used by an unrelated
process at the time C<run_cmd> returns, so it's not useful to send
signals, use as a unique key in data structures and so on.

=back

Example: run C<rm -rf />, redirecting standard input, output and error to
F</dev/null>.

   my $cv = run_cmd [qw(rm -rf /)],
      "<", "/dev/null",
      ">", "/dev/null",
      "2>", "/dev/null";
   $cv->recv and die "d'oh! something survived!"

Example: run F<openssl> and create a self-signed certificate and key,
storing them in C<$cert> and C<$key>. When finished, check the exit status
in the callback and print key and certificate.

   my $cv = run_cmd [qw(openssl req
                     -new -nodes -x509 -days 3650
                     -newkey rsa:2048 -keyout /dev/fd/3
                     -batch -subj /CN=AnyEvent
                    )],
      "<", "/dev/null",
      ">" , \my $cert,
      "3>", \my $key,
      "2>", "/dev/null";

   $cv->cb (sub {
      shift->recv and die "openssl failed";

      print "$key\n$cert\n";
   });

=cut

sub run_cmd {
   my $cmd = shift;

   require POSIX unless $POSIX::VERSION;

   my $cv = AE::cv;

   my %arg;
   my %redir;
   my @exe;

   while (@_) {
      my ($type, $ob) = splice @_, 0, 2;

      my $fd = $type =~ s/^(\d+)// ? $1 : undef;

      if ($type eq ">") {
         $fd = 1 unless defined $fd;

         if (defined eval { fileno $ob }) {
            $redir{$fd} = $ob;
         } elsif (ref $ob) {
            my ($pr, $pw) = AnyEvent::Util::portable_pipe;
            $cv->begin;

            fcntl $pr, AnyEvent::F_SETFD, AnyEvent::FD_CLOEXEC;
            fh_nonblocking $pr, 1;
            my $w; $w = AE::io $pr, 0,
               "SCALAR" eq ref $ob
                  ? sub {
                       defined (sysread $pr, $$ob, 16384, length $$ob
                                and return)
                          or ($! == Errno::EINTR and return);
                       undef $w; $cv->end;
                    }
                  : sub {
                       my $buf;
                       defined (sysread $pr, $buf, 16384
                                and return $ob->($buf))
                          or ($! == Errno::EINTR and return);
                       undef $w; $cv->end;
                       $ob->();
                    }
            ;
            $redir{$fd} = $pw;
         } else {
            push @exe, sub {
               open my $fh, ">", $ob
                  or POSIX::_exit (125);
               $redir{$fd} = $fh;
            };
         }

      } elsif ($type eq "<") {