Fixed bug #34958: --pipe with --record size measured in lines.

src/parallel

@@ -295,49 +295,61 @@ sub spreadstdin {
     # Force the while-loop once if everything was read by header reading
     my $force_one_time_through = 0;
     for my $in (@fhlist) {
-    while(!$force_one_time_through++ or read($in,substr($buf,length $buf,0),$::opt_blocksize)) {
-	# substr above = append to $buf
-	if($::opt_r) {
-	    # Remove empty lines
-	    $buf=~s/^\s*\n//gm;
-	    if(length $buf == 0) {
-		next;
-	    }
-	}
-	if($::opt_regexp) {
-	    if($Global::max_number_of_args) {
-		# -N => (start..*?end){n}
-		while($buf =~ s/((?:$recstart.*?$recend){$Global::max_number_of_args})($recstart.*)$/$2/os) {
-		    write_record_to_pipe(\$header,\$1,$recstart,$recend,length $1);
-		}
+	piperead: while(1) {
+	    if(!$force_one_time_through) {
+		$force_one_time_through++;
+	    } elsif($Global::max_lines) {
+		# Read $Global::max_lines lines
+		eof($in) and last piperead;
+		for(my $t = 0; !eof($in) and 
+		    substr($buf,length $buf,0) = <$in> and $t < $Global::max_lines;
+		    $t++) {}
 	    } else {
-		# Find the last recend-recstart in $buf
-		if($buf =~ s/(.*$recend)($recstart.*?)$/$2/os) {
-		    write_record_to_pipe(\$header,\$1,$recstart,$recend,length $1);
+		# Read a block
+		read($in,substr($buf,length $buf,0),$::opt_blocksize) or last;
+		# substr above = append to $buf
+	    }
+	    if($::opt_r) {
+		# Remove empty lines
+		$buf=~s/^\s*\n//gm;
+		if(length $buf == 0) {
+		    next;
 		}
 	    }
-	} else {
-	    if($Global::max_number_of_args) {
-		# -N => (start..*?end){n}
-		my $i = 0;
-		while(($i = nindex(\$buf,$recendrecstart,$Global::max_number_of_args)) != -1) {
-		    $i += length $recend; # find the actual splitting location
-		    write_record_to_pipe(\$header,\$buf,$recstart,$recend,$i);
-		    substr($buf,0,$i) = "";
+	    if($::opt_regexp) {
+		if($Global::max_number_of_args) {
+		    # -N => (start..*?end){n}
+		    while($buf =~ s/((?:$recstart.*?$recend){$Global::max_number_of_args})($recstart.*)$/$2/os) {
+			write_record_to_pipe(\$header,\$1,$recstart,$recend,length $1);
+		    }
+		} else {
+		    # Find the last recend-recstart in $buf
+		    if($buf =~ s/(.*$recend)($recstart.*?)$/$2/os) {
+			write_record_to_pipe(\$header,\$1,$recstart,$recend,length $1);
+		    }
 		}
 	    } else {
-		# Find the last recend-recstart in $buf
-		my $i = rindex($buf,$recendrecstart);
-		if($i != -1) {
-		    $i += length $recend; # find the actual splitting location
-		    write_record_to_pipe(\$header,\$buf,$recstart,$recend,$i);
-		    substr($buf,0,$i) = "";
+		if($Global::max_number_of_args) {
+		    # -N => (start..*?end){n}
+		    my $i = 0;
+		    while(($i = nindex(\$buf,$recendrecstart,$Global::max_number_of_args)) != -1) {
+			$i += length $recend; # find the actual splitting location
+			write_record_to_pipe(\$header,\$buf,$recstart,$recend,$i);
+			substr($buf,0,$i) = "";
+		    }
+		} else {
+		    # Find the last recend-recstart in $buf
+		    my $i = rindex($buf,$recendrecstart);
+		    if($i != -1) {
+			$i += length $recend; # find the actual splitting location
+			write_record_to_pipe(\$header,\$buf,$recstart,$recend,$i);
+			substr($buf,0,$i) = "";
+		    }
 		}
 	    }
 	}
     }
-}
-
+    
     # If there is anything left in the buffer write it
     substr($buf,0,0) = "";
     write_record_to_pipe(\$header,\$buf,$recstart,$recend,length $buf);

src/parallel.pod

@@ -636,24 +636,28 @@ to see the difference:
 
 =item B<-L> I<max-lines>
 
-Use at most I<max-lines> nonblank input lines per command line.
-Trailing blanks cause an input line to be logically continued on the
-next input line.
+When used with B<--pipe>: Read records of I<max-lines>.
+
+When used otherwise: Use at most I<max-lines> nonblank input lines per
+command line.  Trailing blanks cause an input line to be logically
+continued on the next input line.
 
 B<-L 0> means read one line, but insert 0 arguments on the command
 line.
 
-Implies B<-X> unless B<-m> or B<--xargs> is set.
+Implies B<-X> unless B<-m>, B<--xargs>, or B<--pipe> is set.
 
 
 =item B<--max-lines>[=I<max-lines>]
 
 =item B<-l>[I<max-lines>]
 
-Synonym for the B<-L> option.  Unlike B<-L>, the I<max-lines> argument
-is optional.  If I<max-lines> is not specified, it defaults to one.
-The B<-l> option is deprecated since the POSIX standard specifies
-B<-L> instead.
+When used with B<--pipe>: Read records of I<max-lines>.
+
+When used otherwise: Synonym for the B<-L> option.  Unlike B<-L>, the
+I<max-lines> argument is optional.  If I<max-lines> is not specified,
+it defaults to one.  The B<-l> option is deprecated since the POSIX
+standard specifies B<-L> instead.
 
 B<-l 0> is an alias for B<-l 1>.
 

testsuite/tests-to-run/parallel-local114.sh

@@ -41,4 +41,10 @@ echo "bug #36657: --load does not work with custom ssh"
   export -f ssh; 
   parallel --load=1000% -S "/usr/bin/ssh localhost" echo ::: OK
 
+echo "bug #34958: --pipe with record size measured in lines"
+  seq 10 | parallel --pipe -L 4 cat\;echo FOO
+
+echo "bug #34958: --pipe with record size measured in lines"
+  seq 10 | parallel --pipe -l 4 cat\;echo FOO
+
 EOF

testsuite/wanted-results/parallel-local114

@@ -88,3 +88,18 @@ OK
 OK
 bug #36657: --load does not work with custom ssh
 OK
+bug #34958: --pipe with --record size measured in lines
+1
+2
+3
+4
+FOO
+5
+6
+7
+8
+FOO
+9
+10
+FOO
+FOO