[pve-devel] [RFC common 2/2] fork_worker: use separate pipe for status messages

[Thomas Lamprecht]

We forced line wise flushing of the workers STDOUT and STDERR to
capture the final status (TASK OK/TASK ERROR).
Thus, if the code executed in the worker wanted to flush explicitly,
e.g., when the last output wasn't new line terminated but needed to
reach the users eyes, the parent just ignored that.
This leads to confusing results in CLI handlers using fork_workers.

So remove the buffering logic completely and introduce a separate
pipe for sending the final status.

Signed-off-by: Thomas Lamprecht <t.lamprecht at proxmox.com>
---
 src/PVE/RESTEnvironment.pm | 56 +++++++++++++++++++++++++---------------------
 1 file changed, 30 insertions(+), 26 deletions(-)

diff --git a/src/PVE/RESTEnvironment.pm b/src/PVE/RESTEnvironment.pm
index 0ad6dba..3a2bf8e 100644
--- a/src/PVE/RESTEnvironment.pm
+++ b/src/PVE/RESTEnvironment.pm
@@ -396,6 +396,7 @@ sub fork_worker {
 
     my @psync = POSIX::pipe();
     my @csync = POSIX::pipe();
+    my @ctrlfd = POSIX::pipe() if $sync;
 
     my $node = $self->{nodename};
 
@@ -427,9 +428,11 @@ sub fork_worker {
 	POSIX::setsid();
 
 	POSIX::close ($psync[0]);
+	POSIX::close ($ctrlfd[0]) if $sync;
 	POSIX::close ($csync[1]);
 
 	$outfh = $sync ? $psync[1] : undef;
+	my $resfh = $sync ? $ctrlfd[1] : undef;
 
 	eval {
 	    PVE::INotify::inotify_close();
@@ -450,6 +453,7 @@ sub fork_worker {
 		    if !open(STDIN, "</dev/null");
 
 		$outfh = PVE::Tools::upid_open($upid);
+		$resfh = $outfh;
 	    }
 
 
@@ -499,10 +503,14 @@ sub fork_worker {
 	    chomp $err;
 	    $err =~ s/\n/ /mg;
 	    syslog('err', $err);
-	    print STDERR "TASK ERROR: $err\n";
+	    my $msg = "TASK ERROR: $err\n";
+	    POSIX::write($resfh, $msg, length($msg));
+	    POSIX::close($resfh) if $sync;
 	    POSIX::_exit(-1);
 	} else {
-	    print STDERR "TASK OK\n";
+	    my $msg = "TASK OK\n";
+	    POSIX::write($resfh, $msg, length($msg));
+	    POSIX::close($resfh) if $sync;
 	    POSIX::_exit(0);
 	}
 	kill(-9, $$);
@@ -511,6 +519,7 @@ sub fork_worker {
     # parent
 
     POSIX::close ($psync[1]);
+    POSIX::close ($ctrlfd[1]);
     POSIX::close ($csync[0]);
 
     my $readbuf = '';
@@ -562,7 +571,6 @@ sub fork_worker {
 
     if ($sync) {
 	my $count;
-	my $outbuf = '';
 	my $int_count = 0;
 	eval {
 	    local $SIG{INT} = local $SIG{QUIT} = local $SIG{TERM} = sub {
@@ -591,38 +599,34 @@ sub fork_worker {
 		    }
 		    last if $count == 0; # eof
 
-		    $outbuf .= $readbuf;
-		    while ($outbuf =~ s/^(([^\010\r\n]*)(\r|\n|(\010)+|\r\n))//s) {
-			my $line = $1;
-			my $data = $2;
-			if ($data =~ m/^TASK OK$/) {
-			    # skip
-			} elsif ($data =~ m/^TASK ERROR: (.+)$/) {
-			    print STDERR "$1\n";
-			} else {
-			    print $line;
-			}
-			if ($outfh) {
-			    print $outfh $line;
-			    $outfh->flush();
-			}
-		    }
+		    print $readbuf;
+		    select->flush();
+
+		    # write always to task log (output and control messages)
+		    print $outfh $readbuf;
+		    $outfh->flush();
 		} else {
 		    # some commands daemonize without closing stdout
 		    last if !PVE::ProcFSTools::check_process_running($cpid);
 		}
 	    }
+
+	    # get status (error or OK)
+	    POSIX::read($ctrlfd[0], $readbuf, 4096);
+	    if ($readbuf =~ m/^TASK OK\n?$/) {
+		print $outfh $readbuf;
+	    } elsif ($readbuf =~ m/^TASK ERROR: (.*)\n?$/) {
+		print STDERR "$1\n";
+		print $outfh "\n$readbuf";
+	    } else {
+		die "got unexpected control message: $readbuf\n";
+	    }
+	    $outfh->flush();
 	};
 	my $err = $@;
 
 	POSIX::close($psync[0]);
-
-	if ($outbuf) { # just to be sure
-	    print $outbuf;
-	    if ($outfh) {
-		print $outfh $outbuf;
-	    }
-	}
+	POSIX::close($ctrlfd[0]);
 
 	if ($err) {
 	    $err =~ s/\n/ /mg;
-- 
2.11.0