tange/parallel
1
0
Fork 0
mirror of https://git.savannah.gnu.org/git/parallel.git synced 2024-11-27 16:37:54 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

parallel: --combine-exec implemented.

Browse source
This commit is contained in:
Ole Tange 2023-11-25 19:58:14 +01:00
parent b29cde25f2
commit 0553dbd55c
8 changed files with 412 additions and 223 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

360
src/parallel
View file

@ -299,7 +299,7 @@ sub parcat_script() {
my $file = shift; my $file = shift;
my $output_fd = shift; my $output_fd = shift;
open(my $fh, "<", $file) || do { open(my $fh, "<", $file) || do {
print STDERR "parcat: Cannot open $file\n"; print STDERR "parcat: Cannot open $file: $!\n";
exit(1); exit(1);
}; };
# Remove file when it has been opened # Remove file when it has been opened
@ -597,9 +597,9 @@ sub pipe_shard_setup() {
} }
sub pipe_part_files(@) { sub pipe_part_files(@) {
# Given the bigfile # Given the bigfile:
# find header and split positions # - find header and split positions
# make commands that 'cat's the partial file # - make commands that 'cat's the partial file
# Input: # Input:
# $file = the file to read # $file = the file to read
# Returns: # Returns:
@ -612,7 +612,7 @@ sub pipe_part_files(@) {
::wait_and_exit(255); ::wait_and_exit(255);
} }
my $fh = open_or_exit($file); my $fh = open_or_exit("<",$file);
my $firstlinelen = 0; my $firstlinelen = 0;
if($opt::skip_first_line) { if($opt::skip_first_line) {
my $newline; my $newline;
@ -694,7 +694,7 @@ sub find_split_positions($$$) {
my @pos; my @pos;
my ($recstart,$recend) = recstartrecend(); my ($recstart,$recend) = recstartrecend();
my $recendrecstart = $recend.$recstart; my $recendrecstart = $recend.$recstart;
my $fh = ::open_or_exit($file); my $fh = ::open_or_exit("<",$file);
push(@pos,$skiplen); push(@pos,$skiplen);
for(my $pos = $block+$skiplen; $pos < $size; $pos += $block) { for(my $pos = $block+$skiplen; $pos < $size; $pos += $block) {
my $buf; my $buf;
@ -812,7 +812,7 @@ sub split_positions_for_group_by($$$$) {
my ($file,$size,$block,$header,$firstlinelen) = @_; my ($file,$size,$block,$header,$firstlinelen) = @_;
my @pos; my @pos;
$fh = open_or_exit($file); $fh = open_or_exit("<",$file);
# Set $Global::group_by_column $Global::group_by_perlexpr # Set $Global::group_by_column $Global::group_by_perlexpr
group_by_loop($fh,$opt::recsep); group_by_loop($fh,$opt::recsep);
if($opt::max_args) { if($opt::max_args) {
@ -2130,6 +2130,10 @@ sub options_completion_hash() {
("hgrp|hostgrp|hostgroup|hostgroups[Enable hostgroups on arguments]" ("hgrp|hostgrp|hostgroup|hostgroups[Enable hostgroups on arguments]"
=> \$opt::hostgroups), => \$opt::hostgroups),
"embed[Embed GNU parallel in a shell script]" => \$opt::embed, "embed[Embed GNU parallel in a shell script]" => \$opt::embed,
("filter=s[Only run jobs where filter is true]:filter"
=> \@opt::filter),
"combineexec|combine-exec|combineexecutable|combine-executable=s".
"[Embed GNU parallel in a shell script]" => \$opt::combineexec,
("filter=s[Only run jobs where filter is true]:filter" ("filter=s[Only run jobs where filter is true]:filter"
=> \@opt::filter), => \@opt::filter),
"_parset=s[Generate shell code for parset]" => \$opt::_parset, "_parset=s[Generate shell code for parset]" => \$opt::_parset,
@ -2222,7 +2226,6 @@ sub parse_options(@) {
init_globals(); init_globals();
my @argv_before = @ARGV; my @argv_before = @ARGV;
@ARGV = read_options(); @ARGV = read_options();
# Before changing these line, please read # Before changing these line, please read
# https://www.gnu.org/software/parallel/parallel_design.html#citation-notice # https://www.gnu.org/software/parallel/parallel_design.html#citation-notice
# https://git.savannah.gnu.org/cgit/parallel.git/tree/doc/citation-notice-faq.txt # https://git.savannah.gnu.org/cgit/parallel.git/tree/doc/citation-notice-faq.txt
@ -2367,16 +2370,8 @@ sub parse_options(@) {
push @Global::transfer_files, @opt::transfer_files; push @Global::transfer_files, @opt::transfer_files;
if(%opt::template) { if(%opt::template) {
while (my ($source, $template_name) = each %opt::template) { while (my ($source, $template_name) = each %opt::template) {
if(open(my $tmpl, "<", $source)) {
local $/; # $/ = undef => slurp whole file
my $content = <$tmpl>;
push @Global::template_names, $template_name; push @Global::template_names, $template_name;
push @Global::template_contents, $content; push @Global::template_contents, slurp_or_exit($source);
::debug("tmpl","Name: $template_name\n$content\n");
} else {
::error("Cannot open '$source'.");
wait_and_exit(255);
}
} }
} }
if(not defined $opt::recstart and if(not defined $opt::recstart and
@ -2523,7 +2518,12 @@ sub parse_options(@) {
$Global::ContextReplace = 1; $Global::ContextReplace = 1;
} }
# Deal with ::: :::+ :::: ::::+ and -a +file # Deal with ::: :::+ :::: ::::+ and -a +file
my @ARGV_with_argsep = @ARGV;
@ARGV = read_args_from_command_line(); @ARGV = read_args_from_command_line();
if(defined $opt::combineexec) {
pack_combined_executable(\@argv_before,\@ARGV_with_argsep,\@ARGV);
exit(0);
}
parse_semaphore(); parse_semaphore();
if(defined $opt::eta) { $opt::progress = $opt::eta; } if(defined $opt::eta) { $opt::progress = $opt::eta; }
@ -2652,11 +2652,7 @@ sub parse_options(@) {
delete $ENV{'PARALLEL_ENV'}; delete $ENV{'PARALLEL_ENV'};
if(-e $penv) { if(-e $penv) {
# This is a file/fifo: Replace envvar with content of file # This is a file/fifo: Replace envvar with content of file
open(my $parallel_env, "<", $penv) || $penv = slurp_or_exit($penv);
::die_bug("Cannot read parallel_env from $penv");
local $/; # Put <> in slurp mode
$penv = <$parallel_env>;
close $parallel_env;
} }
# Map \001 to \n to make it easer to quote \n in $PARALLEL_ENV # Map \001 to \n to make it easer to quote \n in $PARALLEL_ENV
$penv =~ s/\001/\n/g; $penv =~ s/\001/\n/g;
@ -3143,12 +3139,7 @@ sub record_env() {
# Record current %ENV-keys in $PARALLEL_HOME/ignored_vars # Record current %ENV-keys in $PARALLEL_HOME/ignored_vars
# Returns: N/A # Returns: N/A
my $ignore_filename = $Global::config_dir . "/ignored_vars"; my $ignore_filename = $Global::config_dir . "/ignored_vars";
if(open(my $vars_fh, ">", $ignore_filename)) { write_or_exit($ignore_filename,map { $_,"\n" } keys %ENV);
print $vars_fh map { $_,"\n" } keys %ENV;
} else {
::error("Cannot write to $ignore_filename.");
::wait_and_exit(255);
}
} }
sub open_joblog() { sub open_joblog() {
@ -3256,24 +3247,17 @@ sub open_joblog() {
} }
if($opt::dryrun) { if($opt::dryrun) {
# Do not write to joblog in a dry-run # Do not write to joblog in a dry-run
if(not open($Global::joblog, ">", "/dev/null")) {
::error("Cannot write to --joblog $opt::joblog.");
::wait_and_exit(255);
}
} elsif($append) { } elsif($append) {
# Append to joblog # Append to joblog
if(not open($Global::joblog, ">>", $opt::joblog)) { $Global::joblog = open_or_exit(">>", $opt::joblog);
::error("Cannot append to --joblog $opt::joblog.");
::wait_and_exit(255);
}
} else { } else {
if($opt::joblog eq "-") { if($opt::joblog eq "-") {
# Use STDOUT as joblog # Use STDOUT as joblog
$Global::joblog = $Global::fh{1}; $Global::joblog = $Global::fh{1};
} elsif(not open($Global::joblog, ">", $opt::joblog)) { } else {
# Overwrite the joblog # Overwrite the joblog
::error("Cannot write to --joblog $opt::joblog."); $Global::joblog = open_or_exit(">", $opt::joblog);
::wait_and_exit(255);
} }
print $Global::joblog print $Global::joblog
join("\t", "Seq", "Host", "Starttime", "JobRuntime", join("\t", "Seq", "Host", "Starttime", "JobRuntime",
@ -3301,11 +3285,7 @@ sub open_json_csv() {
$Global::fh{1} = $fd; $Global::fh{1} = $fd;
$Global::fh{2} = $fd; $Global::fh{2} = $fd;
} elsif($Global::csvsep or $Global::jsonout) { } elsif($Global::csvsep or $Global::jsonout) {
if(not open($Global::csv_fh,">",$opt::results)) { $Global::csv_fh = open_or_exit(">",$opt::results);
::error("Cannot open results file `$opt::results': ".
"$!.");
wait_and_exit(255);
}
} }
} }
} }
@ -3486,7 +3466,7 @@ sub read_options() {
return @ARGV; return @ARGV;
} }
sub arrayindex() { sub arrayindex($$) {
# Similar to Perl's index function, but for arrays # Similar to Perl's index function, but for arrays
# Input: # Input:
# $arr_ref1 = ref to @array1 to search in # $arr_ref1 = ref to @array1 to search in
@ -3894,30 +3874,67 @@ sub enough_file_handles() {
} }
} }
sub open_or_exit($) { sub open_or_exit($$) {
# Open a file name or exit if the file cannot be opened # Open a file name or exit if the file cannot be opened
# Inputs: # Inputs:
# $mode = read:"<" write:">"
# $file = filehandle or filename to open # $file = filehandle or filename to open
# Uses: # Uses:
# $Global::original_stdin # $Global::original_stdin
# Returns: # Returns:
# $fh = file handle to read-opened file # $fh = file handle to opened file
my $mode = shift;
my $file = shift; my $file = shift;
if($file eq "-") { if($file eq "-") {
if($mode eq "<") {
return ($Global::original_stdin || *STDIN); return ($Global::original_stdin || *STDIN);
} else {
return ($Global::original_stderr || *STDERR);
}
} }
if(ref $file eq "GLOB") { if(ref $file eq "GLOB") {
# This is an open filehandle # This is an open filehandle
return $file; return $file;
} }
my $fh = gensym; my $fh = gensym;
if(not open($fh, "<", $file)) { if(not open($fh, $mode, $file)) {
::error("Cannot open input file `$file': No such file or directory."); ::error("Cannot open `$file': $!");
wait_and_exit(255); wait_and_exit(255);
} }
return $fh; return $fh;
} }
sub slurp_or_exit($) {
# Read content of a file or exit if the file cannot be opened
# Inputs:
# $file = filehandle or filename to open
# Returns:
# $content = content as scalar
my $fh = open_or_exit("<",shift);
# $/ = undef => slurp whole file
local $/;
my $content = <$fh>;
close $fh;
return $content;
}
sub write_or_exit(@) {
# Write content to a file or exit if the file cannot be opened
# Inputs:
# $file = filehandle or filename to open
# @content = content to be written
# Returns:
# N/A
my $file = shift;
sub failed {
error("Cannot write to `$file': $!");
wait_and_exit(255);
}
my $fh = open_or_exit(">",$file);
print($fh @_) or failed();
close($fh) or failed();
}
sub set_fh_blocking($) { sub set_fh_blocking($) {
# Set filehandle as blocking # Set filehandle as blocking
# Inputs: # Inputs:
@ -4805,11 +4822,7 @@ sub read_sshloginfile($) {
$in_fh = *STDIN; $in_fh = *STDIN;
$close = 0; $close = 0;
} else { } else {
if(not open($in_fh, "<", $file)) { $in_fh = open_or_exit("<", $file);
# Try the filename
::error("Cannot open $file.");
::wait_and_exit(255);
}
} }
while(<$in_fh>) { while(<$in_fh>) {
chomp; chomp;
@ -5469,8 +5482,7 @@ sub onall($@) {
my %seen; my %seen;
for my $joblog (@joblogs) { for my $joblog (@joblogs) {
# Append to $joblog # Append to $joblog
open(my $fh, "<", $joblog) || my $fh = open_or_exit("<", $joblog);
::die_bug("Cannot open tmp joblog $joblog");
# Skip first line (header); # Skip first line (header);
<$fh>; <$fh>;
print $Global::joblog (<$fh>); print $Global::joblog (<$fh>);
@ -6075,19 +6087,16 @@ sub embed() {
::error("--embed only works if parallel is a readable file"); ::error("--embed only works if parallel is a readable file");
exit(255); exit(255);
} }
if(open(my $fh, "<", $0)) {
# Read the source from $0 # Read the source from $0
my @source = <$fh>; my $source = slurp_or_exit($0);
my $user = $ENV{LOGNAME} || $ENV{USERNAME} || $ENV{USER}; my $user = $ENV{LOGNAME} || $ENV{USERNAME} || $ENV{USER};
my @env_parallel_source = (); my $env_parallel_source;
my $shell = $Global::shell; my $shell = $Global::shell;
$shell =~ s:.*/::; $shell =~ s:.*/::;
for(which("env_parallel.$shell")) { for(which("env_parallel.$shell")) {
-r $_ or next; -r $_ or next;
# Read the source of env_parallel.shellname # Read the source of env_parallel.shellname
open(my $env_parallel_source_fh, $_) || die; $env_parallel_source .= slurp_or_exit($_);
@env_parallel_source = <$env_parallel_source_fh>;
close $env_parallel_source_fh;
last; last;
} }
print "#!$Global::shell print "#!$Global::shell
@ -6137,7 +6146,7 @@ parallel() {
_file_with_GNU_Parallel_source=`mktemp`; _file_with_GNU_Parallel_source=`mktemp`;
!, !,
"cat <<'$randomstring' > \$_file_with_GNU_Parallel_source\n", "cat <<'$randomstring' > \$_file_with_GNU_Parallel_source\n",
@source, $source,
$randomstring,"\n", $randomstring,"\n",
q! q!
# Copy the source code from the file to the fifo # Copy the source code from the file to the fifo
@ -6150,7 +6159,7 @@ parallel() {
perl $_fifo_with_GNU_Parallel_source "$@" perl $_fifo_with_GNU_Parallel_source "$@"
} }
!, !,
@env_parallel_source, $env_parallel_source,
q! q!
# This will call the functions above # This will call the functions above
@ -6162,14 +6171,145 @@ echo $p $y $c $h
echo You can also activate GNU Parallel for interactive use by: echo You can also activate GNU Parallel for interactive use by:
echo . "$0" echo . "$0"
!; !;
} else {
::error("Cannot open $0");
exit(255);
}
::status("Redirect the output to a file and add your changes at the end:", ::status("Redirect the output to a file and add your changes at the end:",
" $0 --embed > new_script"); " $0 --embed > new_script");
} }
sub pack_combined_executable {
my ($before_ref,$with_argsep_ref,$argv_ref) = @_;
my @parallelopts;
my $skip_next;
# Remove '--combine-exec file' from options
for(@{$before_ref}[0..(arrayindex($before_ref,$with_argsep_ref))-1]) {
if (/^--combine-?exec(utable)?$/ || $skip_next) {
# Also skip the filename given to --combine-exec
$skip_next = !$skip_next;
next;
}
push @parallelopts, $_;
}
# From ::: and to end
my @argsep = @{$with_argsep_ref}[($#ARGV+1)..$#$with_argsep_ref];
# The executable is now the first in @ARGV
my $execname = shift @ARGV;
# The rest of @ARGV are options for $execname
my @execopts = @ARGV;
debug("combine",
"Parallel opts: @parallelopts ",
"Executable: $execname ",
"Execopts: @execopts ",
"Argsep: @argsep\n");
# Read the the executable
my $exec = slurp_or_exit(which($execname));
# Read the source of GNU Parallel and the executable
my $parallel = slurp_or_exit($0);
# Remove possibly __END__ from GNU Parallel
$parallel =~ s/^__END__.*//s;
if(-t $Global::original_stderr) {
::status(
"Please be aware that combining GNU Parallel and '$execname'",
"into a combined executable will make the whole executable",
"licensed under GPLv3 (section 5.c).",
"",
"If the license of '$execname' is incompatible with GPLv3,",
"you cannot legally convey copies of the combined executable",
"to others. You can, however, still run them yourself.",
"",
"The combined executable will not have a citation notice,",
"so it is your resposibilty to advice that academic tradition",
"requires the users to cite GNU Parallel.",
""
);
my $input;
do {
::status_no_nl("\nType: 'I agree' and press enter.\n> ");
$input = <STDIN>;
if(not defined $input) {
exit(255);
}
} until($input =~ /I agree/i);
}
write_or_exit($opt::combineexec,
$parallel,
"\n__END__\n",
(map { "$_\0\n" } @parallelopts), "\0\0\n",
$execname, "\0\0\n",
(map { "$_\0\n" } @execopts), "\0\0\n",
(map { "$_\0\n" } @argsep), "\0\0\n",
$exec);
# Set +x permission
chmod 0700, $opt::combineexec;
exit(0);
}
sub unpack_combined_executable {
# If the script is a combined executable,
# it will have stuff in <DATA> (I.e. after __END__)
my $combine_exec = join("",<DATA>);
if(length $combine_exec) {
# Parse the <DATA>
#
# __END__
# Option for GNU Parallel\0\n
# Option for GNU Parallel\0\n
# \0\0\n
# Name of executable\0\0\n
# Option for executable\0\n
# Option for executable\0\n
# \0\0\n
# argsep + args if any\0\n
# argsep + args if any\0\n
# \0\0\n
# <<binary of exec>>
#
# parallel --combine --pipe -j10% --recend '' myscript --myopt myval
# __END__
# --pipe\0\n --pipe
# -j10%\0\n -j10%
# --recend\0\n --recend
# \0\n ''
# \0\0\n end-of-parallel-options
# myscript\0\0\n myscript
# --myopt\0\n --myopt
# myval\0\n myval
# \0\0\n end-of-myscript-options
# \0\0\n no argsep
# <<binary of myscript>>
#
# parallel --combine -j10% myscript :::
# __END__
# -j10%\0\n
# \0\0\n end-of-parallel-options
# myscript\0\0\n
# \0\0\n end-of-myscript-options
# :::\0\n
# \0\0\n
# <<binary of myscript>>
my ($opts,$execname,$execopts,$argsep,$exec) =
split /\0\0\n/,$combine_exec,5;
# Make a tmpdir with a file called $execname
local %ENV;
$ENV{TMPDIR} ||= "/tmp";
my $dir = File::Temp::tempdir($ENV{'TMPDIR'} . "/parXXXXX", CLEANUP => 1);
my $script = $dir."/".$execname;
write_or_exit($script,$exec);
# Set +x permission
chmod 0700, $script;
# Mark it for unlinking later
$Global::unlink{$script}++;
$Global::unlink{$dir}++;
# pass the options for GNU Parallel
my @opts = split /\0\n/, $opts;
my @execopts = split /\0\n/, $execopts;
if(length $argsep) {
# Only add argsep if set
unshift(@ARGV, split(/\0\n/,$argsep));
}
unshift(@ARGV,@opts,$script,@execopts);
}
}
sub __GENERIC_COMMON_FUNCTION__() {} sub __GENERIC_COMMON_FUNCTION__() {}
@ -6245,15 +6385,11 @@ sub size_of_block_dev() {
# Returns: # Returns:
# $size = in bytes, undef if error # $size = in bytes, undef if error
my $blockdev = shift; my $blockdev = shift;
if(open(my $fh, "<", $blockdev)) { my $fh = open_or_exit("<", $blockdev);
seek($fh,0,2) || ::die_bug("cannot seek $blockdev"); seek($fh,0,2) || ::die_bug("cannot seek $blockdev");
my $size = tell($fh); my $size = tell($fh);
close $fh; close $fh;
return $size; return $size;
} else {
::error("cannot open $blockdev");
wait_and_exit(255);
}
} }
sub qqx(@) { sub qqx(@) {
@ -8014,14 +8150,9 @@ sub compute_max_loadavg($) {
} elsif (-f $loadspec) { } elsif (-f $loadspec) {
$Global::max_load_file = $loadspec; $Global::max_load_file = $loadspec;
$Global::max_load_file_last_mod = (stat($Global::max_load_file))[9]; $Global::max_load_file_last_mod = (stat($Global::max_load_file))[9];
if(open(my $in_fh, "<", $Global::max_load_file)) { $load = $self->compute_max_loadavg(
my $opt_load_file = join("",<$in_fh>); ::slurp_or_exit($Global::max_load_file)
close $in_fh; );
$load = $self->compute_max_loadavg($opt_load_file);
} else {
::error("Cannot open $loadspec.");
::wait_and_exit(255);
}
} else { } else {
::error("Parsing of --load failed."); ::error("Parsing of --load failed.");
::die_usage(); ::die_usage();
@ -8358,19 +8489,13 @@ sub user_requested_processes($) {
if(defined $opt_P) { if(defined $opt_P) {
if (-f $opt_P) { if (-f $opt_P) {
$Global::max_procs_file = $opt_P; $Global::max_procs_file = $opt_P;
if(open(my $in_fh, "<", $Global::max_procs_file)) { my $opt_P_file = ::slurp_or_exit($Global::max_procs_file);
my $opt_P_file = join("",<$in_fh>);
close $in_fh;
if($opt_P_file !~ /\S/) { if($opt_P_file !~ /\S/) {
::warning_once("$Global::max_procs_file is empty. ". ::warning_once("$Global::max_procs_file is empty. ".
"Treated as 100%"); "Treated as 100%");
$opt_P_file = "100%"; $opt_P_file = "100%";
} }
$processes = $self->user_requested_processes($opt_P_file); $processes = $self->user_requested_processes($opt_P_file);
} else {
::error("Cannot open $opt_P.");
::wait_and_exit(255);
}
} else { } else {
if($opt_P eq "0") { if($opt_P eq "0") {
# -P 0 = infinity (or at least close) # -P 0 = infinity (or at least close)
@ -8561,20 +8686,14 @@ sub sct_gnu_linux($) {
for($thread = 0; for($thread = 0;
-r "$prefix/cpu$thread/topology/physical_package_id"; -r "$prefix/cpu$thread/topology/physical_package_id";
$thread++) { $thread++) {
open(my $fh,"<", $socket{slurp_or_exit(
"$prefix/cpu$thread/topology/physical_package_id") "$prefix/cpu$thread/topology/physical_package_id")}++;
|| die;
$socket{<$fh>}++;
close $fh;
} }
for($thread = 0; for($thread = 0;
-r "$prefix/cpu$thread/topology/thread_siblings"; -r "$prefix/cpu$thread/topology/thread_siblings";
$thread++) { $thread++) {
open(my $fh,"<", $sibiling{slurp_or_exit(
"$prefix/cpu$thread/topology/thread_siblings") "$prefix/cpu$thread/topology/thread_siblings")}++;
|| die;
$sibiling{<$fh>}++;
close $fh;
} }
$cpu->{'sockets'} = keys %socket; $cpu->{'sockets'} = keys %socket;
$cpu->{'cores'} = keys %sibiling; $cpu->{'cores'} = keys %sibiling;
@ -9417,21 +9536,9 @@ sub openoutputfiles($) {
$errname = "$out.err"; $errname = "$out.err";
$seqname = "$out.seq"; $seqname = "$out.seq";
} }
my $seqfhw; ::write_or_exit($seqname, $self->seq());
if(not open($seqfhw, "+>", $seqname)) { $outfhw = ::open_or_exit("+>", $outname);
::error("Cannot write to `$seqname'."); $errfhw = ::open_or_exit("+>", $errname);
::wait_and_exit(255);
}
print $seqfhw $self->seq();
close $seqfhw;
if(not open($outfhw, "+>", $outname)) {
::error("Cannot write to `$outname'.");
::wait_and_exit(255);
}
if(not open($errfhw, "+>", $errname)) {
::error("Cannot write to `$errname'.");
::wait_and_exit(255);
}
$self->set_fh(1,"unlink",""); $self->set_fh(1,"unlink","");
$self->set_fh(2,"unlink",""); $self->set_fh(2,"unlink","");
if($opt::sqlworker) { if($opt::sqlworker) {
@ -9530,8 +9637,7 @@ sub grouped($) {
# Re-open the file for reading # Re-open the file for reading
# so fdw can be closed seperately # so fdw can be closed seperately
# and fdr can be seeked seperately (for --line-buffer) # and fdr can be seeked seperately (for --line-buffer)
open(my $fdr,"<", $self->fh($fdno,'name')) || my $fdr = ::open_or_exit("<", $self->fh($fdno,'name'));
::die_bug("fdr: Cannot open ".$self->fh($fdno,'name'));
$self->set_fh($fdno,'r',$fdr); $self->set_fh($fdno,'r',$fdr);
# Unlink if not debugging # Unlink if not debugging
$Global::debug or ::rm($self->fh($fdno,"unlink")); $Global::debug or ::rm($self->fh($fdno,"unlink"));
@ -10367,9 +10473,7 @@ sub sshlogin_wrap($) {
if(-r $_ and not -d) { if(-r $_ and not -d) {
# Read as environment definition bug #44041 # Read as environment definition bug #44041
# TODO parse this # TODO parse this
my $fh = ::open_or_exit($_); $Global::envdef = ::slurp_or_exit($_);
$Global::envdef = join("",<$fh>);
close $fh;
} }
} }
if(grep { /^_$/ } @vars) { if(grep { /^_$/ } @vars) {
@ -10549,11 +10653,11 @@ sub fill_templates($) {
@{$self->{'commandline'}{'template_names'}}; @{$self->{'commandline'}{'template_names'}};
::debug("tmpl","Names: @template_name\n"); ::debug("tmpl","Names: @template_name\n");
for(my $i = 0; $i <= $#template_name; $i++) { for(my $i = 0; $i <= $#template_name; $i++) {
open(my $fh, ">", $template_name[$i]) || die; ::write_or_exit
print $fh $self->{'commandline'}-> ($template_name[$i],
$self->{'commandline'}->
replace_placeholders([$self->{'commandline'} replace_placeholders([$self->{'commandline'}
{'template_contents'}[$i]],0,0); {'template_contents'}[$i]],0,0));
close $fh;
} }
if($opt::cleanup) { if($opt::cleanup) {
$self->add_rm(@template_name); $self->add_rm(@template_name);
@ -11036,7 +11140,7 @@ sub interactive_start($) {
my $answer; my $answer;
::status_no_nl("$command ?..."); ::status_no_nl("$command ?...");
do{ do{
open(my $tty_fh, "<", "/dev/tty") || ::die_bug("interactive-tty"); my $tty_fh = ::open_or_exit("<","/dev/tty");
$answer = <$tty_fh>; $answer = <$tty_fh>;
close $tty_fh; close $tty_fh;
# Sometime we get an empty string (not even \n) # Sometime we get an empty string (not even \n)
@ -14590,7 +14694,7 @@ sub get_alias($) {
# local $/ needed if -0 set # local $/ needed if -0 set
local $/ = "\n"; local $/ = "\n";
if(-r $alias_file) { if(-r $alias_file) {
open(my $in, "<", $alias_file) || die; my $in = ::open_or_exit("<",$alias_file);
push @urlalias, <$in>; push @urlalias, <$in>;
close $in; close $in;
} }
@ -15211,7 +15315,9 @@ $Global::max_slot_number = $opt::session;
package main; package main;
sub main() { sub main() {
unpack_combined_executable();
save_stdin_stdout_stderr(); save_stdin_stdout_stderr();
save_original_signal_handler(); save_original_signal_handler();
parse_options(); parse_options();
@ -15229,7 +15335,7 @@ sub main() {
my @input_source_fh; my @input_source_fh;
if($opt::pipepart) { if($opt::pipepart) {
if($opt::tee) { if($opt::tee) {
@input_source_fh = map { open_or_exit($_) } @opt::a; @input_source_fh = map { open_or_exit("<",$_) } @opt::a;
# Remove the first: It will be the file piped. # Remove the first: It will be the file piped.
shift @input_source_fh; shift @input_source_fh;
if(not @input_source_fh and not $opt::pipe) { if(not @input_source_fh and not $opt::pipe) {
@ -15237,10 +15343,10 @@ sub main() {
} }
} else { } else {
# -a is used for data - not for command line args # -a is used for data - not for command line args
@input_source_fh = map { open_or_exit($_) } "/dev/null"; @input_source_fh = map { open_or_exit("<",$_) } "/dev/null";
} }
} else { } else {
@input_source_fh = map { open_or_exit($_) } @opt::a; @input_source_fh = map { open_or_exit("<",$_) } @opt::a;
if(not @input_source_fh and not $opt::pipe) { if(not @input_source_fh and not $opt::pipe) {
@input_source_fh = (*STDIN); @input_source_fh = (*STDIN);
} }

48
src/parallel.pod
View file

@ -536,9 +536,9 @@ Shorthand for B<--delimiter '\0'>.
See also: B<--delimiter> See also: B<--delimiter>
=item B<--arg-file> I<input-file> (alpha testing) =item B<--arg-file> I<input-file> (beta testing)
=item B<-a> I<input-file> (alpha testing) =item B<-a> I<input-file> (beta testing)
Use I<input-file> as input source. Use I<input-file> as input source.
@ -841,6 +841,50 @@ https://perldoc.perl.org/perlre.html
See also: B<--csv> B<{>I<n>B<}> B<--trim> B<--link> See also: B<--csv> B<{>I<n>B<}> B<--trim> B<--link>
=item B<--combineexec> I<name> (alpha testing)
=item B<--combine-executable> I<name> (alpha testing)
Combine GNU B<parallel> with another program into a single executable.
Let us say you have developed I<myprg> which takes a single
argument. You do not want to parallelize it yourself.
You could write a wrapper that uses GNU B<parallel> called B<myparprg>:
#!/bin/sh
parallel myprg ::: "$@"
But for others to use this, they need to install: GNU B<parallel>,
B<myprg>, and B<myparprg>.
It would be easier to install if all could be packed into a single
executable.
If B<myprg> is written in shell, you can use B<--embed>.
If B<myprg> is a binary you can use B<--combineexec>.
Here we use B<gzip> as example:
parallel --combineexec pargzip gzip -9 :::
You can now do:
./pargzip foo bar baz
If you want to pass options to B<gzip> you can do:
parallel --combineexec pargzip gzip
Followed by:
./pargzip -1 ::: foo bar baz
See also: B<--embed> B<--shebang> B<--shebang-wrap>
=item B<--compress> =item B<--compress>
Compress temporary files. Compress temporary files.

34
testsuite/tests-to-run/parallel-local-3s.sh
View file

@ -8,6 +8,29 @@
# Each should be taking 3-10s and be possible to run in parallel # Each should be taking 3-10s and be possible to run in parallel
# I.e.: No race conditions, no logins # I.e.: No race conditions, no logins
par_combineexec() {
combineexec() {
stderr=$(mktemp)
parallel --combineexec "$combo" "$@" 2>"$stderr"
# Redirected stderr should give no output
cat "$stderr"
rm "$stderr"
}
combo=$(mktemp)
echo '### Check that "--pipe -k" works'
combineexec -j 2 -k --pipe wc
seq 920000 | "$combo"
echo '### Check that "-k" is kept'
combineexec -k bash -c :::
"$combo" 'sleep 0.$RANDOM; echo 1' 'sleep 0.$RANDOM; echo 2' 'sleep 0.$RANDOM; echo 3'
echo '### Check that "--tagstring {1}" is kept'
combineexec --tagstring {1} -k perl -e :::
"$combo" 'print("1\n")' 'print("2\n")' 'print("3\n")'
}
par__argfile_plus() { par__argfile_plus() {
tmp=$(mktemp -d) tmp=$(mktemp -d)
( (
@ -483,17 +506,6 @@ par_maxargs() {
(echo line 1;echo line 1;echo line 2) | parallel -k --max-args 2 echo (echo line 1;echo line 1;echo line 2) | parallel -k --max-args 2 echo
} }
par_totaljob_repl() {
echo '{##} bug #45841: Replacement string for total no of jobs'
parallel -k --plus echo {##} ::: {a..j};
parallel -k 'echo {= $::G++ > 3 and ($_=$Global::JobQueue->total_jobs());=}' ::: {1..10}
parallel -k -N7 --plus echo {#} {##} ::: {1..14}
parallel -k -N7 --plus echo {#} {##} ::: {1..15}
parallel -k -S 8/: -X --plus echo {#} {##} ::: {1..15}
parallel -k --plus --delay 0.01 -j 10 'sleep 2; echo {0#}/{##}:{0%}' ::: {1..5} ::: {1..4}
}
par_jobslot_repl() { par_jobslot_repl() {
echo 'bug #46232: {%} with --bar/--eta/--shuf or --halt xx% broken' echo 'bug #46232: {%} with --bar/--eta/--shuf or --halt xx% broken'

11
testsuite/tests-to-run/parallel-local-race02.sh
View file

@ -42,6 +42,17 @@ ctrlz_should_suspend_children() {
} }
ctrlz_should_suspend_children ctrlz_should_suspend_children
par_totaljob_repl() {
echo '{##} bug #45841: Replacement string for total no of jobs'
parallel -k --plus echo {##} ::: {a..j};
parallel -k 'echo {= $::G++ > 3 and ($_=$Global::JobQueue->total_jobs());=}' ::: {1..10}
parallel -k -N7 --plus echo {#} {##} ::: {1..14}
parallel -k -N7 --plus echo {#} {##} ::: {1..15}
parallel -k -S 8/: -X --plus echo {#} {##} ::: {1..15}
parallel -k --plus --delay 0.01 -j 10 'sleep 2; echo {0#}/{##}:{0%}' ::: {1..5} ::: {1..4}
}
par_semaphore() { par_semaphore() {
echo '### Test if parallel invoked as sem will run parallel --semaphore' echo '### Test if parallel invoked as sem will run parallel --semaphore'
sem --id as_sem -u -j2 'echo job1a 1; sleep 3; echo job1b 3' sem --id as_sem -u -j2 'echo job1a 1; sleep 3; echo job1b 3'

8
testsuite/wanted-results/parallel-local-0.3s
View file

@ -970,10 +970,10 @@ par_sem_quote ### sem --quote should not add empty argument
par_sem_quote echo par_sem_quote echo
par_sem_quote par_sem_quote
par_shellcompletion ### --shellcompletion par_shellcompletion ### --shellcompletion
par_shellcompletion 70960cbdbc411e041161ae228f029d70 - par_shellcompletion 1952015cc0c85c44d82d13e2673f4b28 -
par_shellcompletion 70960cbdbc411e041161ae228f029d70 - par_shellcompletion 1952015cc0c85c44d82d13e2673f4b28 -
par_shellcompletion aa125ab894780611a20bad4a52d7a58d - par_shellcompletion c23b2094e510526c322ea3af89b4c682 -
par_shellcompletion aa125ab894780611a20bad4a52d7a58d - par_shellcompletion c23b2094e510526c322ea3af89b4c682 -
par_slow_pipe_regexp ### bug #53718: --pipe --regexp -N blocks par_slow_pipe_regexp ### bug #53718: --pipe --regexp -N blocks
par_slow_pipe_regexp This should take a few ms, but took more than 2 hours par_slow_pipe_regexp This should take a few ms, but took more than 2 hours
par_slow_pipe_regexp 0 1 1 par_slow_pipe_regexp 0 1 1

70
testsuite/wanted-results/parallel-local-3s
View file

@ -331,6 +331,22 @@ par_children_receive_sig parallel: Warning: This job was killed because it timed
par_children_receive_sig parallel: Warning: show_signals '' par_children_receive_sig parallel: Warning: show_signals ''
par_children_receive_sig Got INT par_children_receive_sig Got INT
par_children_receive_sig Got TERM par_children_receive_sig Got TERM
par_combineexec ### Check that "--pipe -k" works
par_combineexec 165668 165668 1048571
par_combineexec 149796 149796 1048572
par_combineexec 149796 149796 1048572
par_combineexec 149796 149796 1048572
par_combineexec 149796 149796 1048572
par_combineexec 149796 149796 1048572
par_combineexec 5352 5352 37464
par_combineexec ### Check that "-k" is kept
par_combineexec 1
par_combineexec 2
par_combineexec 3
par_combineexec ### Check that "--tagstring {1}" is kept
par_combineexec print("1\n") 1
par_combineexec print("2\n") 2
par_combineexec print("3\n") 3
par_delay ### Test --delay par_delay ### Test --delay
par_delay More than 3.3 secs: OK par_delay More than 3.3 secs: OK
par_delay_halt_soon bug #59893: --halt soon doesn't work with --delay par_delay_halt_soon bug #59893: --halt soon doesn't work with --delay
@ -996,60 +1012,6 @@ par_test_delimiter ### Test : as delimiter. This can be confusing for uptime ie.
par_test_delimiter a par_test_delimiter a
par_test_delimiter b par_test_delimiter b
par_test_delimiter c par_test_delimiter c
par_totaljob_repl {##} bug #45841: Replacement string for total no of jobs
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 1
par_totaljob_repl 2
par_totaljob_repl 3
par_totaljob_repl 4
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 1 2
par_totaljob_repl 2 2
par_totaljob_repl 1 3
par_totaljob_repl 2 3
par_totaljob_repl 3 3
par_totaljob_repl 1 15
par_totaljob_repl 2 14
par_totaljob_repl 3 14
par_totaljob_repl 4 14
par_totaljob_repl 5 14
par_totaljob_repl 6 14
par_totaljob_repl 7 14
par_totaljob_repl 8 14
par_totaljob_repl 01/20:01
par_totaljob_repl 02/20:02
par_totaljob_repl 03/20:03
par_totaljob_repl 04/20:04
par_totaljob_repl 05/20:05
par_totaljob_repl 06/20:06
par_totaljob_repl 07/20:07
par_totaljob_repl 08/20:08
par_totaljob_repl 09/20:09
par_totaljob_repl 10/20:10
par_totaljob_repl 11/20:01
par_totaljob_repl 12/20:02
par_totaljob_repl 13/20:03
par_totaljob_repl 14/20:04
par_totaljob_repl 15/20:05
par_totaljob_repl 16/20:06
par_totaljob_repl 17/20:07
par_totaljob_repl 18/20:08
par_totaljob_repl 19/20:09
par_totaljob_repl 20/20:10
par_wrong_slot_rpl_resume ### bug #47644: Wrong slot number replacement when resuming par_wrong_slot_rpl_resume ### bug #47644: Wrong slot number replacement when resuming
par_wrong_slot_rpl_resume 1 0 par_wrong_slot_rpl_resume 1 0
par_wrong_slot_rpl_resume 2 1 par_wrong_slot_rpl_resume 2 1

54
testsuite/wanted-results/parallel-local-race02
View file

@ -1502,3 +1502,57 @@ par_testhalt soon done 70% false job 17
par_testhalt soon done 70% false parallel: This job finished: par_testhalt soon done 70% false parallel: This job finished:
par_testhalt soon done 70% false echo job 17; sleep 7.5; exit 7 par_testhalt soon done 70% false echo job 17; sleep 7.5; exit 7
par_testhalt soon done 70% false 20 par_testhalt soon done 70% false 20
par_totaljob_repl {##} bug #45841: Replacement string for total no of jobs
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 1
par_totaljob_repl 2
par_totaljob_repl 3
par_totaljob_repl 4
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 10
par_totaljob_repl 1 2
par_totaljob_repl 2 2
par_totaljob_repl 1 3
par_totaljob_repl 2 3
par_totaljob_repl 3 3
par_totaljob_repl 1 15
par_totaljob_repl 2 14
par_totaljob_repl 3 14
par_totaljob_repl 4 14
par_totaljob_repl 5 14
par_totaljob_repl 6 14
par_totaljob_repl 7 14
par_totaljob_repl 8 14
par_totaljob_repl 01/20:01
par_totaljob_repl 02/20:02
par_totaljob_repl 03/20:03
par_totaljob_repl 04/20:04
par_totaljob_repl 05/20:05
par_totaljob_repl 06/20:06
par_totaljob_repl 07/20:07
par_totaljob_repl 08/20:08
par_totaljob_repl 09/20:09
par_totaljob_repl 10/20:10
par_totaljob_repl 11/20:01
par_totaljob_repl 12/20:02
par_totaljob_repl 13/20:03
par_totaljob_repl 14/20:04
par_totaljob_repl 15/20:05
par_totaljob_repl 16/20:06
par_totaljob_repl 17/20:07
par_totaljob_repl 18/20:08
par_totaljob_repl 19/20:09
par_totaljob_repl 20/20:10

6
testsuite/wanted-results/parallel-local9
View file

@ -199,7 +199,7 @@ seq 0 7 | $XAP -kN3 echo {1} {2} {3}
echo '### Test :::: on nonexistent' echo '### Test :::: on nonexistent'
### Test :::: on nonexistent ### Test :::: on nonexistent
stdout $XAP -k echo {1} {2} {3} :::: nonexistent stdout $XAP -k echo {1} {2} {3} :::: nonexistent
parallel: Error: Cannot open input file `nonexistent': No such file or directory. parallel: Error: Cannot open `nonexistent': No such file or directory
echo '### Test :::: two files' echo '### Test :::: two files'
### Test :::: two files ### Test :::: two files
$XAP -k echo {1} {2} :::: <(seq 1 10) <(seq 5 15) $XAP -k echo {1} {2} :::: <(seq 1 10) <(seq 5 15)
@ -313,9 +313,9 @@ b a
echo '### Multiple -a: nonexistent' echo '### Multiple -a: nonexistent'
### Multiple -a: nonexistent ### Multiple -a: nonexistent
stdout $XAP -kv echo {2} {1} :::: nonexist nonexist2 stdout $XAP -kv echo {2} {1} :::: nonexist nonexist2
parallel: Error: Cannot open input file `nonexist': No such file or directory. parallel: Error: Cannot open `nonexist': No such file or directory
stdout $XAP -kv -a nonexist -a nonexist2 echo {2} {1} stdout $XAP -kv -a nonexist -a nonexist2 echo {2} {1}
parallel: Error: Cannot open input file `nonexist': No such file or directory. parallel: Error: Cannot open `nonexist': No such file or directory
echo '### Test {#.}' echo '### Test {#.}'
### Test {#.} ### Test {#.}
$XAP -kv -a <(echo a-noext) -a <(echo b-withext.extension) -a <(echo c-ext.gif) echo {3.} {2.} {1.} $XAP -kv -a <(echo a-noext) -a <(echo b-withext.extension) -a <(echo c-ext.gif) echo {3.} {2.} {1.}