Merge pull request #22 from cfconrad/fix_race_on_register_process_v2

Fix race on `open3/fork` and `session->register($pid)` call -- second attempt

Author: foursixnine

lib/Mojo/IOLoop/ReadWriteProcess.pm

@@ -120,7 +120,7 @@ sub _open {
   my ($self, @args) = @_;
   $self->_diag('Execute: ' . (join ', ', map { "'$_'" } @args)) if DEBUG;
 
-  $self->session->enable;
+  $self->on(collect_status => \&_open_collect_status);
 
   my ($wtr, $rdr, $err);
   $err = gensym;
@@ -130,7 +130,6 @@ sub _open {
   $self->process_id($pid);
 
   # Defered collect of return status and removal of pidfile
-  $self->on(collect_status => \&_open_collect_status);
 
   return $self unless $self->set_pipes();
 
@@ -220,7 +219,6 @@ sub _fork {
   # Separated handles that could be used for internal comunication.
   my ($channel_in, $channel_out);
 
-  $self->session->enable;
 
   if ($self->set_pipes) {
     $input_pipe = IO::Pipe->new()
@@ -397,7 +395,11 @@ sub exit_status {
 sub restart {
   $_[0]->is_running ? $_[0]->stop->start : $_[0]->start;
 }
-sub is_running { $_[0]->process_id ? kill 0 => $_[0]->process_id : 0; }
+sub is_running {
+    my ($self) = shift;
+    $self->session->consume_collected_info;
+    $self->process_id ? kill 0 => $self->process_id : 0;
+}
 
 sub write_pidfile {
   my ($self, $pidfile) = @_;
@@ -472,6 +474,8 @@ sub start {
 
   $self->session->enable_subreaper if $self->subreaper;
   $self->_status(undef);
+  $self->session->enable;
+
 
   if ($self->code) {
     $self->_fork($self->code, @args);

lib/Mojo/IOLoop/ReadWriteProcess/Session.pm

@@ -21,6 +21,7 @@ has subreaper      => 0;
 has collect_status => 1;
 has orphans        => sub { {} };
 has process_table  => sub { {} };
+has collected_info => sub { [] };
 has 'handler';
 
 my $singleton;
@@ -52,7 +53,7 @@ sub enable {
         $singleton->emit('SIG_CHLD');
         return unless $singleton->collect_status;
         while ((my $pid = waitpid(-1, WNOHANG)) > 0) {
-          $singleton->collect($pid => $? => $!);
+          $singleton->add_collected_info($pid, $?, $!);
         }
       }
     });
@@ -66,7 +67,7 @@ sub _collect {
 }
 
 sub collect {
-  my ($errno, $status, $pid) = (pop, pop, pop);
+  my ($self, $pid, $status, $errno) = @_;
   if ($singleton->resolve($pid)) {
     $singleton->_collect($pid => $status => $errno);
     $singleton->emit(collected => $singleton->resolve($pid));
@@ -80,6 +81,17 @@ sub collect {
   return $singleton;
 }
 
+sub consume_collected_info {
+    while(my $i = shift @{$singleton->collected_info}) {
+        $singleton->collect(@$i) 
+    }
+}
+
+sub add_collected_info {
+    shift;
+    push @{$singleton->collected_info}, [@_];
+}
+
 # Use as $pid => Mojo::IOLoop::ReadWriteProcess
 sub register {
   my ($process, $pid) = (pop, pop);
@@ -121,7 +133,7 @@ sub contains {
   $singleton->all->grep(sub { $_->pid eq $pid })->size == 1;
 }
 
-sub reset { @{+shift}{qw(events orphans process_table)} = ({}, {}, {}) }
+sub reset { @{+shift}{qw(events orphans process_table collected_info handler)} = ({}, {}, {}, [], undef) }
 
 # XXX: This should be replaced by PR_GET_CHILD_SUBREAPER
 sub disable_subreaper {

t/01_run.t

@@ -301,6 +301,13 @@ subtest 'process execute()' => sub {
   $p->stop();
 };
 
+subtest 'process(execute =>"/usr/bin/true")' => sub {
+  use Mojo::IOLoop::ReadWriteProcess qw(process);
+  plan skip_all => "Missing '/usr/bin/true'" unless -e '/usr/bin/true';
+
+  is(process(execute => '/usr/bin/true')->quirkiness(1)->start()->wait_stop()->exit_status(), 0, 'Simple exec of "/usr/bin/true" return 0');
+};
+
 subtest 'process code()' => sub {
   use Mojo::IOLoop::ReadWriteProcess;
   use IO::Select;
@@ -560,4 +567,15 @@ subtest 'process_args' => sub {
   is($p->read_all_stdout(), "0$/1$/2$/3$/", '2) Args given as arrayref.');
 };
 
+subtest 'process in process' => sub {
+    my $p = process(sub {
+        is( process(execute => '/usr/bin/true')->quirkiness(1)->start()->wait_stop()->exit_status(), 0, 'process(execute) from process(code) -- retval check true');
+        is( process(execute => '/usr/bin/false')->quirkiness(1)->start()->wait_stop()->exit_status(), 1, 'process(execute) from process(code) -- retval check false');
+        is( process(sub { print 'sub-sub-process'})->start()->wait_stop()->read_all_stdout, 'sub-sub-process', 'process(code) works from process(code)');
+        print 'DONE';
+    })->start()->wait_stop();
+
+    is ($p->read_all_stdout(), 'DONE', "Use ReadWriteProcess inside of ReadWriteProcess(code=>'')");
+};
+
 done_testing;