tange/parallel
1
0
Fork 0
mirror of https://git.savannah.gnu.org/git/parallel.git synced 2024-12-22 20:57:53 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

parallel: --files/--results --line-buffer fix.

Browse source
This commit is contained in:
Ole Tange 2016-09-22 23:54:38 +02:00
parent 5829f512c1
commit a4ddcbf8bc
1 changed files with 60 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

111
src/parallel
View file

@ -8107,11 +8107,11 @@ sub print {
next;
}
::debug("print", "File descriptor $fdno (", $self->fh($fdno,"name"), "):\n");
if($opt::files) {
$self->files_print($fdno,$in_fh,$out_fd);
} elsif($opt::linebuffer) {
if($opt::linebuffer) {
# Line buffered print out
$self->linebuffer_print($fdno,$in_fh,$out_fd);
} elsif($opt::files) {
$self->files_print($fdno,$in_fh,$out_fd);
} elsif($opt::tag or defined $opt::tagstring) {
$self->tag_print($fdno,$in_fh,$out_fd);
} else {
@ -8137,20 +8137,18 @@ sub files_print {
my $self = shift;
my ($fdno,$in_fh,$out_fd) = @_;
if($opt::group and defined $self->{'exitstatus'}) {
# If the job is dead: close printing fh. Needed for --compress
close $self->fh($fdno,"w");
if($? and $opt::compress) {
::error($opt::compress_program." failed.");
$self->set_exitstatus(255);
}
if($opt::compress) {
# Kill the decompressor which will not be needed
CORE::kill "TERM", $self->fh($fdno,"rpid");
}
close $in_fh;
# If the job is dead: close printing fh. Needed for --compress
close $self->fh($fdno,"w");
if($? and $opt::compress) {
::error($opt::compress_program." failed.");
$self->set_exitstatus(255);
}
if($opt::compress) {
# Kill the decompressor which will not be needed
CORE::kill "TERM", $self->fh($fdno,"rpid");
}
close $in_fh;
if($opt::pipe and $self->virgin()) {
# Nothing was printed to this job:
# cleanup unused tmp files if --files was set
@ -8189,46 +8187,57 @@ sub linebuffer_print {
}
}
}
# This seek will clear EOF
seek $in_fh, tell($in_fh), 0;
# The read is non-blocking: The $in_fh is set to non-blocking.
# 32768 --tag = 5.1s
# 327680 --tag = 4.4s
# 1024000 --tag = 4.4s
# 3276800 --tag = 4.3s
# 10240000 --tag = 4.3s
# 32768000 --tag = 4.7s
my $outputlength = 0;
while(read($in_fh,substr($$partial,length $$partial),3276800)) {
# Append to $$partial
# Find the last \n or \r
my $i = (::rindex64($partial,"\n")+1) || (::rindex64($partial,"\r")+1);
if($i) {
# One or more complete lines were found
$outputlength += $i;
if($opt::tag or defined $opt::tagstring) {
# Replace ^ with $tag within the full line
my $tag = $self->tag();
substr($$partial,0,$i) =~ s/^/$tag/gm;
# Length of partial line has changed: Find the last \n/\r again
$i = (::rindex64($partial,"\n")+1) || (::rindex64($partial,"\r")+1);
if($opt::files or $opt::results) {
if($fdno == 1 and not $self->fh($fdno,"printed")) {
print $out_fd $self->tag(),$self->fh($fdno,"name"),"\n";
$self->set_fh($fdno,"printed",1);
}
} else {
# This seek will clear EOF
seek $in_fh, tell($in_fh), 0;
my $outputlength = 0;
# The read is non-blocking: The $in_fh is set to non-blocking.
# 32768 --tag = 5.1s
# 327680 --tag = 4.4s
# 1024000 --tag = 4.4s
# 3276800 --tag = 4.3s
# 10240000 --tag = 4.3s
# 32768000 --tag = 4.7s
while(read($in_fh,substr($$partial,length $$partial),3276800)) {
# Append to $$partial
# Find the last \n or \r
my $i = (::rindex64($partial,"\n")+1) || (::rindex64($partial,"\r")+1);
if($i) {
# One or more complete lines were found
$outputlength += $i;
if($opt::tag or defined $opt::tagstring) {
# Replace ^ with $tag within the full line
my $tag = $self->tag();
substr($$partial,0,$i) =~ s/^/$tag/gm;
# Length of partial line has changed: Find the last \n/\r again
$i = (::rindex64($partial,"\n")+1) || (::rindex64($partial,"\r")+1);
}
# Print up to and including the last \n
print $out_fd substr($$partial,0,$i);
# Remove the printed part
substr($$partial,0,$i) = "";
}
# Print up to and including the last \n
print $out_fd substr($$partial,0,$i);
# Remove the printed part
substr($$partial,0,$i) = "";
}
$self->add_returnsize($outputlength);
}
$self->add_returnsize($outputlength);
if(defined $self->{'exitstatus'}) {
# If the job is dead: print the remaining partial line
# read remaining
$self->add_returnsize(length $$partial);
if($$partial and ($opt::tag or defined $opt::tagstring)) {
my $tag = $self->tag();
$$partial =~ s/^/$tag/gm;
if($opt::files or $opt::results) {
$self->add_returnsize(-s $self->fh($fdno,"name"));
} else {
# If the job is dead: print the remaining partial line
# read remaining
$self->add_returnsize(length $$partial);
if(length $$partial and ($opt::tag or defined $opt::tagstring)) {
my $tag = $self->tag();
$$partial =~ s/^/$tag/gm;
}
print $out_fd $$partial;
}
print $out_fd $$partial;
# Release the memory
undef $$partial;
if($self->fh($fdno,"rpid") and CORE::kill 0, $self->fh($fdno,"rpid")) {