Merge branch 'master' of ssh://git.sv.gnu.org/srv/git/parallel

This commit is contained in:
Ole Tange 2017-02-13 23:21:29 +01:00
commit 5ea9f57ba7
24 changed files with 575 additions and 214 deletions

View file

@ -59,4 +59,60 @@ upload:
pushd; \
sudo cp /usr/local/bin/parallel /usr/local/bin/parallel-$(YYYYMMDD)
alphaupload:
# Copy of upload
eval `gpg-agent --daemon`
# make sure $YYYYMMDD is set
echo $(YYYYMMDD) | grep 20
cp parallel-$(YYYYMMDD).tar.bz2 parallel-latest.tar.bz2
cp doc/parallel.directive parallel-$(YYYYMMDD).tar.bz2.directive
perl -i -pe "s/20\d\d\d\d\d\d/$(YYYYMMDD)/" parallel-*.tar.*directive
gpg --clearsign --yes parallel-$(YYYYMMDD).tar.bz2.directive
cp doc/parallel.latest.directive parallel-latest.tar.bz2.directive
perl -i -pe "s/20\d\d\d\d\d\d/$(YYYYMMDD)/" parallel-latest.tar.*directive
gpg --clearsign --yes parallel-latest.tar.bz2.directive
(echo '#!/bin/bash'; \
echo; \
echo "# To check the signature run:"; \
echo "# echo | gpg"; \
echo "# gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve parallel-$(YYYYMMDD).tar.bz2.sig"; \
echo; \
echo "echo | gpg 2>/dev/null"; \
echo 'gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve $$0'; \
echo 'exit $$?' ; \
echo; \
gpg -ab -o - parallel-$(YYYYMMDD).tar.bz2; \
) > parallel-$(YYYYMMDD).tar.bz2.sig
(echo '#!/bin/bash'; \
echo; \
echo "# To check the signature run:"; \
echo "# echo | gpg"; \
echo "# gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve parallel-latest.tar.bz2.sig"; \
echo; \
echo "echo | gpg 2>/dev/null"; \
echo 'gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve $$0'; \
echo 'exit $$?' ; \
echo; \
gpg -ab -o - parallel-$(YYYYMMDD).tar.bz2; \
) > parallel-latest.tar.bz2.sig
gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve parallel-latest.tar.bz2.sig
gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve parallel-$(YYYYMMDD).tar.bz2.sig
../ftpsync/src/ftpsync parallel-$(YYYYMMDD).tar.bz2{,.sig,*asc} ftp://ftp-upload.gnu.org/incoming/alpha/
# This can take 7 minutes
pushd /tmp; \
rm -rf /tmp/parallel-$(YYYYMMDD)*; \
while ! wget http://alpha.gnu.org/gnu/parallel/parallel-$(YYYYMMDD).tar.bz2 ; do sleep 2; done; \
tar xjvf parallel-$(YYYYMMDD).tar.bz2; \
cd parallel-$(YYYYMMDD); \
./configure; \
make -j && sudo make -j install; \
pushd; \
sudo cp /usr/local/bin/parallel /usr/local/bin/parallel-$(YYYYMMDD)
EXTRA_DIST = CITATION CREDITS

View file

@ -807,6 +807,62 @@ upload:
pushd; \
sudo cp /usr/local/bin/parallel /usr/local/bin/parallel-$(YYYYMMDD)
alphaupload:
# Copy of upload
eval `gpg-agent --daemon`
# make sure $YYYYMMDD is set
echo $(YYYYMMDD) | grep 20
cp parallel-$(YYYYMMDD).tar.bz2 parallel-latest.tar.bz2
cp doc/parallel.directive parallel-$(YYYYMMDD).tar.bz2.directive
perl -i -pe "s/20\d\d\d\d\d\d/$(YYYYMMDD)/" parallel-*.tar.*directive
gpg --clearsign --yes parallel-$(YYYYMMDD).tar.bz2.directive
cp doc/parallel.latest.directive parallel-latest.tar.bz2.directive
perl -i -pe "s/20\d\d\d\d\d\d/$(YYYYMMDD)/" parallel-latest.tar.*directive
gpg --clearsign --yes parallel-latest.tar.bz2.directive
(echo '#!/bin/bash'; \
echo; \
echo "# To check the signature run:"; \
echo "# echo | gpg"; \
echo "# gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve parallel-$(YYYYMMDD).tar.bz2.sig"; \
echo; \
echo "echo | gpg 2>/dev/null"; \
echo 'gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve $$0'; \
echo 'exit $$?' ; \
echo; \
gpg -ab -o - parallel-$(YYYYMMDD).tar.bz2; \
) > parallel-$(YYYYMMDD).tar.bz2.sig
(echo '#!/bin/bash'; \
echo; \
echo "# To check the signature run:"; \
echo "# echo | gpg"; \
echo "# gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve parallel-latest.tar.bz2.sig"; \
echo; \
echo "echo | gpg 2>/dev/null"; \
echo 'gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve $$0'; \
echo 'exit $$?' ; \
echo; \
gpg -ab -o - parallel-$(YYYYMMDD).tar.bz2; \
) > parallel-latest.tar.bz2.sig
gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve parallel-latest.tar.bz2.sig
gpg --auto-key-locate keyserver --keyserver-options auto-key-retrieve parallel-$(YYYYMMDD).tar.bz2.sig
../ftpsync/src/ftpsync parallel-$(YYYYMMDD).tar.bz2{,.sig,*asc} ftp://ftp-upload.gnu.org/incoming/alpha/
# This can take 7 minutes
pushd /tmp; \
rm -rf /tmp/parallel-$(YYYYMMDD)*; \
while ! wget http://alpha.gnu.org/gnu/parallel/parallel-$(YYYYMMDD).tar.bz2 ; do sleep 2; done; \
tar xjvf parallel-$(YYYYMMDD).tar.bz2; \
cd parallel-$(YYYYMMDD); \
./configure; \
make -j && sudo make -j install; \
pushd; \
sudo cp /usr/local/bin/parallel /usr/local/bin/parallel-$(YYYYMMDD)
# Tell versions [3.59,3.63) of GNU make to not export all variables.
# Otherwise a system limit (for SysV at least) may be exceeded.
.NOEXPORT:

5
NEWS
View file

@ -1,3 +1,8 @@
20170206alpha
* --tee introduced.
20170122
* sql now uses a temporary credentials file for MySQL to avoid

12
README
View file

@ -44,9 +44,9 @@ document.
Full installation of GNU Parallel is as simple as:
wget http://ftpmirror.gnu.org/parallel/parallel-20170122.tar.bz2
bzip2 -dc parallel-20170122.tar.bz2 | tar xvf -
cd parallel-20170122
wget http://ftpmirror.gnu.org/parallel/parallel-20170206.tar.bz2
bzip2 -dc parallel-20170206.tar.bz2 | tar xvf -
cd parallel-20170206
./configure && make && sudo make install
@ -55,9 +55,9 @@ Full installation of GNU Parallel is as simple as:
If you are not root you can add ~/bin to your path and install in
~/bin and ~/share:
wget http://ftpmirror.gnu.org/parallel/parallel-20170122.tar.bz2
bzip2 -dc parallel-20170122.tar.bz2 | tar xvf -
cd parallel-20170122
wget http://ftpmirror.gnu.org/parallel/parallel-20170206.tar.bz2
bzip2 -dc parallel-20170206.tar.bz2 | tar xvf -
cd parallel-20170206
./configure --prefix=$HOME && make && make install
Or if your system lacks 'make' you can simply copy src/parallel

20
configure vendored
View file

@ -1,6 +1,6 @@
#! /bin/sh
# Guess values for system-dependent variables and create Makefiles.
# Generated by GNU Autoconf 2.69 for parallel 20170122.
# Generated by GNU Autoconf 2.69 for parallel 20170206.
#
# Report bugs to <bug-parallel@gnu.org>.
#
@ -579,8 +579,8 @@ MAKEFLAGS=
# Identity of this package.
PACKAGE_NAME='parallel'
PACKAGE_TARNAME='parallel'
PACKAGE_VERSION='20170122'
PACKAGE_STRING='parallel 20170122'
PACKAGE_VERSION='20170206'
PACKAGE_STRING='parallel 20170206'
PACKAGE_BUGREPORT='bug-parallel@gnu.org'
PACKAGE_URL=''
@ -1214,7 +1214,7 @@ if test "$ac_init_help" = "long"; then
# Omit some internal or obsolete options to make the list less imposing.
# This message is too long to be a string in the A/UX 3.1 sh.
cat <<_ACEOF
\`configure' configures parallel 20170122 to adapt to many kinds of systems.
\`configure' configures parallel 20170206 to adapt to many kinds of systems.
Usage: $0 [OPTION]... [VAR=VALUE]...
@ -1281,7 +1281,7 @@ fi
if test -n "$ac_init_help"; then
case $ac_init_help in
short | recursive ) echo "Configuration of parallel 20170122:";;
short | recursive ) echo "Configuration of parallel 20170206:";;
esac
cat <<\_ACEOF
@ -1357,7 +1357,7 @@ fi
test -n "$ac_init_help" && exit $ac_status
if $ac_init_version; then
cat <<\_ACEOF
parallel configure 20170122
parallel configure 20170206
generated by GNU Autoconf 2.69
Copyright (C) 2012 Free Software Foundation, Inc.
@ -1374,7 +1374,7 @@ cat >config.log <<_ACEOF
This file contains any messages produced by compilers while
running configure, to aid debugging if configure makes a mistake.
It was created by parallel $as_me 20170122, which was
It was created by parallel $as_me 20170206, which was
generated by GNU Autoconf 2.69. Invocation command line was
$ $0 $@
@ -2237,7 +2237,7 @@ fi
# Define the identity of the package.
PACKAGE='parallel'
VERSION='20170122'
VERSION='20170206'
cat >>confdefs.h <<_ACEOF
@ -2880,7 +2880,7 @@ cat >>$CONFIG_STATUS <<\_ACEOF || ac_write_fail=1
# report actual input values of CONFIG_FILES etc. instead of their
# values after options handling.
ac_log="
This file was extended by parallel $as_me 20170122, which was
This file was extended by parallel $as_me 20170206, which was
generated by GNU Autoconf 2.69. Invocation command line was
CONFIG_FILES = $CONFIG_FILES
@ -2942,7 +2942,7 @@ _ACEOF
cat >>$CONFIG_STATUS <<_ACEOF || ac_write_fail=1
ac_cs_config="`$as_echo "$ac_configure_args" | sed 's/^ //; s/[\\""\`\$]/\\\\&/g'`"
ac_cs_version="\\
parallel config.status 20170122
parallel config.status 20170206
configured by $0, generated by GNU Autoconf 2.69,
with options \\"\$ac_cs_config\\"

View file

@ -1,4 +1,4 @@
AC_INIT([parallel], [20170122], [bug-parallel@gnu.org])
AC_INIT([parallel], [20170206], [bug-parallel@gnu.org])
AM_INIT_AUTOMAKE([-Wall -Werror foreign])
AC_CONFIG_HEADERS([config.h])
AC_CONFIG_FILES([

View file

@ -230,9 +230,10 @@ DISTCLEANFILES = parallel.1 env_parallel.1 sem.1 sql.1 niceload.1 \
parallel_tutorial.pdf parallel_design.pdf parallel_alternatives.pdf \
parcat.pdf
EXTRA_DIST = parallel sem sql niceload parcat env_parallel \
env_parallel.bash env_parallel.zsh env_parallel.fish env_parallel.ksh \
env_parallel.pdksh env_parallel.csh env_parallel.tcsh \
sem.pod parallel.pod env_parallel.pod niceload.pod parallel_tutorial.pod \
parallel_design.pod parallel_alternatives.pod \
$(DISTCLEANFILES)
EXTRA_DIST = parallel sem sql niceload parcat env_parallel \
env_parallel.ash env_parallel.bash env_parallel.csh \
env_parallel.dash env_parallel.fish env_parallel.ksh \
env_parallel.pdksh env_parallel.tcsh env_parallel.zsh \
sem.pod parallel.pod env_parallel.pod niceload.pod \
parallel_tutorial.pod parallel_design.pod \
parallel_alternatives.pod $(DISTCLEANFILES)

View file

@ -152,7 +152,7 @@ NROFF = nroff
MANS = $(man_MANS)
DATA = $(doc_DATA)
am__tagged_files = $(HEADERS) $(SOURCES) $(TAGS_FILES) $(LISP)
am__DIST_COMMON = $(srcdir)/Makefile.in README
am__DIST_COMMON = $(srcdir)/Makefile.in
DISTFILES = $(DIST_COMMON) $(DIST_SOURCES) $(TEXINFOS) $(EXTRA_DIST)
ACLOCAL = @ACLOCAL@
AMTAR = @AMTAR@
@ -229,10 +229,10 @@ target_alias = @target_alias@
top_build_prefix = @top_build_prefix@
top_builddir = @top_builddir@
top_srcdir = @top_srcdir@
bin_SCRIPTS = parallel sql niceload parcat \
env_parallel env_parallel.bash env_parallel.csh \
env_parallel.dash env_parallel.fish env_parallel.ksh \
env_parallel.pdksh env_parallel.sh env_parallel.tcsh \
bin_SCRIPTS = parallel sql niceload parcat env_parallel \
env_parallel.ash env_parallel.bash env_parallel.csh \
env_parallel.dash env_parallel.fish env_parallel.ksh \
env_parallel.pdksh env_parallel.sh env_parallel.tcsh \
env_parallel.zsh
@DOCUMENTATION_TRUE@man_MANS = parallel.1 env_parallel.1 sem.1 sql.1 niceload.1 \
@ -262,12 +262,13 @@ DISTCLEANFILES = parallel.1 env_parallel.1 sem.1 sql.1 niceload.1 \
parallel_tutorial.pdf parallel_design.pdf parallel_alternatives.pdf \
parcat.pdf
EXTRA_DIST = parallel sem sql niceload parcat env_parallel \
env_parallel.bash env_parallel.zsh env_parallel.fish env_parallel.ksh \
env_parallel.pdksh env_parallel.csh env_parallel.tcsh \
sem.pod parallel.pod env_parallel.pod niceload.pod parallel_tutorial.pod \
parallel_design.pod parallel_alternatives.pod \
$(DISTCLEANFILES)
EXTRA_DIST = parallel sem sql niceload parcat env_parallel \
env_parallel.ash env_parallel.bash env_parallel.csh \
env_parallel.dash env_parallel.fish env_parallel.ksh \
env_parallel.pdksh env_parallel.tcsh env_parallel.zsh \
sem.pod parallel.pod env_parallel.pod niceload.pod \
parallel_tutorial.pod parallel_design.pod \
parallel_alternatives.pod $(DISTCLEANFILES)
all: all-am

View file

@ -18,28 +18,41 @@
# or write to the Free Software Foundation, Inc., 51 Franklin St,
# Fifth Floor, Boston, MA 02110-1301 USA
GREPQ="grep >/dev/null 2>/dev/null"
grepq() {
# grep -q for systems without -q
grep >/dev/null 2>/dev/null "$@"
}
installer() {
source="$1"
script="$2"
into="$3"
if grepq $script $into; then
true already installed
else
echo $source \`which $script\` >> $into
fi
}
while test $# -gt 0; do
key="$1"
case $key in
-i|--install)
eval $GREPQ env_parallel.bash $HOME/.bashrc ||
echo '. `which env_parallel.bash`' >> $HOME/.bashrc
eval $GREPQ env_parallel.zsh $HOME/.zshenv ||
echo '. `which env_parallel.zsh`' >> $HOME/.zshenv
installer . env_parallel.bash $HOME/.bashrc
installer . env_parallel.zsh $HOME/.zshenv
installer source env_parallel.ksh $HOME/.kshrc
echo $SHELL | grepq /pdksh &&
installer . env_parallel.pdksh $HOME/.profile
echo $SHELL | grepq /ash &&
installer . env_parallel.ash $HOME/.profile
echo $SHELL | grepq /dash &&
installer . env_parallel.dash $HOME/.profile
installer source env_parallel.csh $HOME/.cshrc
installer source env_parallel.tcsh $HOME/.tcshrc
mkdir -p $HOME/.config/fish
eval $GREPQ env_parallel.fish $HOME/.config/fish/config.fish ||
grepq env_parallel.fish $HOME/.config/fish/config.fish ||
echo '. (which env_parallel.fish)' >> $HOME/.config/fish/config.fish
eval $GREPQ env_parallel.ksh $HOME/.kshrc ||
echo 'source `which env_parallel.ksh`' >> $HOME/.kshrc
eval $GREPQ env_parallel.pdksh $HOME/.profile ||
echo '. `which env_parallel.pdksh`' >> $HOME/.profile
eval $GREPQ env_parallel.csh $HOME/.cshrc ||
echo 'source `which env_parallel.csh`' >> $HOME/.cshrc
eval $GREPQ env_parallel.tcsh $HOME/.tcshrc ||
echo 'source `which env_parallel.tcsh`' >> $HOME/.tcshrc
echo 'Installed env_parallel in: '
echo " " $HOME/.bashrc
echo " " $HOME/.zshenv
@ -83,6 +96,14 @@ pdksh: Put this in $HOME/.profile: source `which env_parallel.pdksh`
E.g. by doing: echo '. `which env_parallel.pdksh`' >> $HOME/.profile
Supports: aliases, functions, variables, arrays
ash: Put this in $HOME/.profile: . `which env_parallel.ash`
E.g. by doing: echo '. `which env_parallel.ash`' >> $HOME/.profile
Supports: aliases, variables
dash: Put this in $HOME/.profile: . `which env_parallel.dash`
E.g. by doing: echo '. `which env_parallel.dash`' >> $HOME/.profile
Supports: aliases, variables
csh: Put this in $HOME/.cshrc: source `which env_parallel.csh`
E.g. by doing: echo 'source `which env_parallel.csh`' >> $HOME/.cshrc
Supports: aliases, variables, arrays with no special chars

View file

@ -81,7 +81,7 @@ Same as GNU B<parallel>.
=head1 SUPPORTED SHELLS
=head2 ash
=head2 Ash
B<--env> is supported to export only the variable, or alias with the
given name. Multiple B<--env>s can be given.
@ -120,11 +120,10 @@ E.g. by doing:
=item arrays
ash does not support arrays.
Arrays are not supported by Ash.
=back
=head2 Bash
B<--env> is supported to export only the variable, alias, function, or
@ -228,7 +227,7 @@ Not supported by B<csh>.
=back
=head2 dash
=head2 Dash
B<--env> is supported to export only the variable, or alias with the
given name. Multiple B<--env>s can be given.

View file

@ -24,7 +24,7 @@
use strict;
use Getopt::Long;
$Global::progname="niceload";
$Global::version = 20170123;
$Global::version = 20170206;
Getopt::Long::Configure("bundling","require_order");
get_options_from_array(\@ARGV) || die_usage();
if($opt::version) {

View file

@ -51,8 +51,17 @@ my @command = @ARGV;
my @input_source_fh;
if($opt::pipepart) {
# -a is used for data - not for command line args
@input_source_fh = map { open_or_exit($_) } "/dev/null";
if($opt::tee) {
@input_source_fh = map { open_or_exit($_) } @opt::a;
# Remove the first: It will be the file piped.
shift @input_source_fh;
if(not @input_source_fh and not $opt::pipe) {
@input_source_fh = (*STDIN);
}
} else {
# -a is used for data - not for command line args
@input_source_fh = map { open_or_exit($_) } "/dev/null";
}
} else {
@input_source_fh = map { open_or_exit($_) } @opt::a;
if(not @input_source_fh and not $opt::pipe) {
@ -77,6 +86,65 @@ if($opt::skip_first_line) {
set_input_source_header();
if($opt::filter_hosts and (@opt::sshlogin or @opt::sshloginfile)) {
# Parallel check all hosts are up. Remove hosts that are down
filter_hosts();
}
if($opt::nonall or $opt::onall) {
onall(\@input_source_fh,@command);
wait_and_exit(min(undef_as_zero($Global::exitstatus),254));
}
# TODO --transfer foo/./bar --cleanup
# multiple --transfer and --basefile with different /./
$Global::JobQueue = JobQueue->new(
\@command,\@input_source_fh,$Global::ContextReplace,
$number_of_args,\@Global::transfer_files,\@Global::ret_files);
if($opt::pipepart) {
pipepart_setup();
} elsif($opt::pipe and $opt::tee) {
pipe_tee_setup();
}
if($opt::eta or $opt::bar or $opt::shuf or $Global::halt_pct) {
# Count the number of jobs or shuffle all jobs
# before starting any.
# Must be done after ungetting any --pipepart jobs.
$Global::JobQueue->total_jobs();
}
# Compute $Global::max_jobs_running
# Must be done after ungetting any --pipepart jobs.
max_jobs_running();
init_run_jobs();
my $sem;
if($Global::semaphore) {
$sem = acquire_semaphore();
}
$SIG{TERM} = \&start_no_new_jobs;
start_more_jobs();
if($opt::tee) {
# All jobs must be running in parallel for --tee
$Global::start_no_new_jobs = 1;
} elsif($opt::pipe and not $opt::pipepart) {
spreadstdin();
}
::debug("init", "Start draining\n");
drain_job_queue();
::debug("init", "Done draining\n");
reaper();
::debug("init", "Done reaping\n");
::debug("init", "Cleaning\n");
if($Global::semaphore) {
$sem->release();
}
cleanup();
::debug("init", "Halt\n");
halt();
sub set_input_source_header {
if($opt::header and not $opt::pipe) {
# split with colsep or \t
@ -96,7 +164,10 @@ sub set_input_source_header {
for my $s (split /$delimiter/o, $line) {
::debug("init", "Colname: '$s'");
# Replace {colname} with {2}
for(@command,@Global::ret_files,@Global::transfer_files) {
for(@command,@Global::ret_files,@Global::transfer_files,
$opt::tagstring, $opt::workdir, $opt::results, $opt::retries) {
# Skip if undefined
$_ or next;
s:\{$s(|/|//|\.|/\.)\}:\{$id$1\}:g;
# {=header1 ... =} => {=1 ... =}
s:$left $s (.*?) $right:$l$id$1$r:gx;
@ -114,122 +185,102 @@ sub set_input_source_header {
}
}
if($opt::filter_hosts and (@opt::sshlogin or @opt::sshloginfile)) {
# Parallel check all hosts are up. Remove hosts that are down
filter_hosts();
}
if($opt::nonall or $opt::onall) {
onall(\@input_source_fh,@command);
wait_and_exit(min(undef_as_zero($Global::exitstatus),254));
}
# TODO --transfer foo/./bar --cleanup
# multiple --transfer and --basefile with different /./
$Global::JobQueue = JobQueue->new(
\@command,\@input_source_fh,$Global::ContextReplace,
$number_of_args,\@Global::transfer_files,\@Global::ret_files);
if($opt::pipepart) {
if(not $opt::blocksize) {
# --blocksize with 10 jobs per jobslot
$opt::blocksize = -10;
}
if($opt::roundrobin) {
# --blocksize with 1 job per jobslot
$opt::blocksize = -1;
}
if($opt::blocksize < 0) {
my $size = 0;
# Compute size of -a
for(@opt::a) {
if(-f $_) {
$size += -s $_;
} elsif(-b $_) {
$size += size_of_block_dev($_);
} else {
::error("$_ is neither a file nor a block device");
wait_and_exit(255);
}
}
sub max_jobs_running {
if(not $Global::max_jobs_running) {
# Compute $Global::max_jobs_running
$Global::dummy_jobs = 1;
for my $sshlogin (values %Global::host) {
$sshlogin->max_jobs_running();
}
$Global::max_jobs_running or
::die_bug("Global::max_jobs_running not set");
# Run in total $job_slots*(- $blocksize) jobs
# Set --blocksize = size / no of proc / (- $blocksize)
$Global::blocksize = 1 +
int($size / $Global::max_jobs_running / -$opt::blocksize);
}
@Global::cat_partials = map { pipe_part_files($_) } @opt::a;
# Unget the empty arg as many times as there are parts
$Global::JobQueue->{'commandlinequeue'}{'arg_queue'}->unget(
map { [Arg->new("\0")] } @Global::cat_partials
);
}
if($opt::eta or $opt::bar or $opt::shuf or $Global::halt_pct) {
# Count the number of jobs or shuffle all jobs
# before starting any.
# Must be done after ungetting any --pipepart jobs.
$Global::JobQueue->total_jobs();
}
# Compute $Global::max_jobs_running
# Must be done after ungetting any --pipepart jobs.
for my $sshlogin (values %Global::host) {
$sshlogin->max_jobs_running();
return $Global::max_jobs_running;
}
init_run_jobs();
my $sem;
if($Global::semaphore) {
$sem = acquire_semaphore();
}
$SIG{TERM} = \&start_no_new_jobs;
start_more_jobs();
if($opt::pipe and not $opt::pipepart) {
spreadstdin();
}
::debug("init", "Start draining\n");
drain_job_queue();
::debug("init", "Done draining\n");
reaper();
::debug("init", "Done reaping\n");
if($opt::pipe and @opt::a) {
for my $job (@Global::tee_jobs) {
::rm($job->fh(2,"name"));
$job->set_fh(2,"name","");
$job->print();
::rm($job->fh(1,"name"));
}
}
::debug("init", "Cleaning\n");
if($Global::semaphore) {
$sem->release();
}
cleanup();
::debug("init", "Halt\n");
if($opt::halt and $Global::halt_when ne "never") {
if(not defined $Global::halt_exitstatus) {
if($Global::halt_pct) {
$Global::halt_exitstatus =
::ceil($Global::total_failed / $Global::total_started * 100);
} elsif($Global::halt_count) {
$Global::halt_exitstatus =
::min(undef_as_zero($Global::total_failed),101);
sub halt {
if($opt::halt and $Global::halt_when ne "never") {
if(not defined $Global::halt_exitstatus) {
if($Global::halt_pct) {
$Global::halt_exitstatus =
::ceil($Global::total_failed / $Global::total_started * 100);
} elsif($Global::halt_count) {
$Global::halt_exitstatus =
::min(undef_as_zero($Global::total_failed),101);
}
}
wait_and_exit($Global::halt_exitstatus);
} else {
wait_and_exit(min(undef_as_zero($Global::exitstatus),101));
}
wait_and_exit($Global::halt_exitstatus);
} else {
wait_and_exit(min(undef_as_zero($Global::exitstatus),101));
}
sub __PIPE_MODE__ {}
sub pipepart_setup {
if($opt::tee) {
# Prepend each command with
# cat file
my $cat_string = "cat ".::shell_quote_scalar($opt::a[0]);
for(1..max_jobs_running()) {
push @Global::cat_prepends, $cat_string;
}
} else {
if(not $opt::blocksize) {
# --blocksize with 10 jobs per jobslot
$opt::blocksize = -10;
}
if($opt::roundrobin) {
# --blocksize with 1 job per jobslot
$opt::blocksize = -1;
}
if($opt::blocksize < 0) {
my $size = 0;
# Compute size of -a
for(@opt::a) {
if(-f $_) {
$size += -s $_;
} elsif(-b $_) {
$size += size_of_block_dev($_);
} else {
::error("$_ is neither a file nor a block device");
wait_and_exit(255);
}
}
# Run in total $job_slots*(- $blocksize) jobs
# Set --blocksize = size / no of proc / (- $blocksize)
$Global::dummy_jobs = 1;
$Global::blocksize = 1 +
int($size / max_jobs_running() / -$opt::blocksize);
}
@Global::cat_prepends = map { pipe_part_files($_) } @opt::a;
# Unget the empty arg as many times as there are parts
$Global::JobQueue->{'commandlinequeue'}{'arg_queue'}->unget(
map { [Arg->new("\0")] } @Global::cat_prepends
);
}
}
sub pipe_tee_setup {
# mkfifo t1..5
my @fifos;
for(1..max_jobs_running()) {
push @fifos, tmpfifo();
}
# cat foo | tee t1 t2 t3 t4 t5 > /dev/null
if(not fork()){
# Let tee inheirit our stdin
# and redirect stdout to null
open STDOUT, ">","/dev/null";
exec "tee",@fifos;
}
# cat t1 | grep 1
# cat t2 | grep 2
# cat t3 | grep 3
# cat t4 | grep 4
# cat t5 | grep 5
# Remove the tmpfifo as soon as it is open
@Global::cat_prepends = map { "(rm $_;cat) < $_" } @fifos;
}
sub pipe_part_files {
# Input:
@ -245,12 +296,12 @@ sub pipe_part_files {
my $header = find_header(\$buf,open_or_exit($file));
# find positions
my @pos = find_split_positions($file,$Global::blocksize,length $header);
# Make @cat_partials
my @cat_partials = ();
# Make @cat_prepends
my @cat_prepends = ();
for(my $i=0; $i<$#pos; $i++) {
push @cat_partials, cat_partial($file, 0, length($header), $pos[$i], $pos[$i+1]);
push @cat_prepends, cat_partial($file, 0, length($header), $pos[$i], $pos[$i+1]);
}
return @cat_partials;
return @cat_prepends;
}
sub find_header {
@ -341,22 +392,26 @@ sub cat_partial {
# $file = the file to read
# ($start, $end, [$start2, $end2, ...]) = start byte, end byte
# Returns:
# Efficient perl command to copy $start..$end, $start2..$end2, ... to stdout
# Efficient command to copy $start..$end, $start2..$end2, ... to stdout
my($file, @start_end) = @_;
my($start, $i);
# Convert start_end to start_len
my @start_len = map { if(++$i % 2) { $start = $_; } else { $_-$start } } @start_end;
my $script = spacefree(0,
q{
while(@ARGV) {
sysseek(STDIN,shift,0) || die;
$left = shift;
while($read = sysread(STDIN,$buf, $left > 131072 ? 131072 : $left)){
$left -= $read;
syswrite(STDOUT,$buf);
}
}
});
my @start_len = map {
if(++$i % 2) { $start = $_; } else { $_-$start }
} @start_end;
my $script = spacefree
(0,
q{
while(@ARGV) {
sysseek(STDIN,shift,0) || die;
$left = shift;
while($read =
sysread(STDIN,$buf, $left > 131072 ? 131072 : $left)){
$left -= $read;
syswrite(STDOUT,$buf);
}
}
});
return "<". shell_quote_scalar($file) .
" perl -e '$script' @start_len";
}
@ -376,6 +431,15 @@ sub spreadstdin {
# %Global::running
# Returns: N/A
if($opt::tee) {
# Spawn all jobs
# read a record
# Write record to all jobs
if(not $Global::JobQueue->empty()) {
::error("--tee requres --jobs to be higher. Try --jobs 0.");
}
}
my $buf = "";
my ($recstart,$recend) = recstartrecend();
my $recendrecstart = $recend.$recstart;
@ -597,24 +661,38 @@ sub nindex {
push @robin_queue, (sort { $a->seq() <=> $b->seq() }
values %Global::running);
}
do {
$written = 0;
if($opt::keeporder) {
for my $job (@robin_queue) {
if($job->block_length() > 0) {
$written += $job->non_blocking_write();
} else {
$job->set_block($header_ref,$buffer_ref,
$endpos,$recstart,$recend);
$job->set_block($header_ref,$buffer_ref,$endpos,$recstart,$recend);
$block_passed = 1;
$job->set_virgin(0);
$written += $job->non_blocking_write();
last;
}
}
if($written) {
$sleep = $sleep/1.5+0.001;
}
} while($written and not $block_passed);
} else {
do {
$written = 0;
for my $job (@robin_queue) {
if($job->block_length() > 0) {
$written += $job->non_blocking_write();
} else {
$job->set_block($header_ref,$buffer_ref,
$endpos,$recstart,$recend);
$block_passed = 1;
$job->set_virgin(0);
$written += $job->non_blocking_write();
last;
}
}
if($written) {
$sleep = $sleep/1.5+0.001;
}
} while($written and not $block_passed);
}
$sleep = ::reap_usleep($sleep);
}
return $written;
@ -730,6 +808,7 @@ sub write_record_to_pipe {
if($endpos == 0) { return 0; }
if(vec($Global::job_already_run,$chunk_number,1)) { return 1; }
if($opt::roundrobin) {
# Write the block to one of the already running jobs
return round_robin_write($header_ref,$buffer_ref,$recstart,$recend,$endpos);
}
# If no virgin found, backoff
@ -961,6 +1040,7 @@ sub options_hash {
"cat" => \$opt::cat,
"fifo" => \$opt::fifo,
"pipepart|pipe-part" => \$opt::pipepart,
"tee" => \$opt::tee,
"hgrp|hostgrp|hostgroup|hostgroups" => \$opt::hostgroups,
);
}
@ -1137,6 +1217,11 @@ sub parse_options {
map { (0..9,"a".."z","A".."Z")[rand(62)] } (1..50);
push @ARGV, $Global::arg_sep, "\0";
}
if(defined $opt::tee) {
if(not defined $opt::jobs) {
$opt::jobs = 0;
}
}
if(defined $opt::tty) {
# Defaults for --tty: -j1 -u
# Can be overridden with -jXXX -g
@ -1276,7 +1361,7 @@ sub check_invalid_option_combinations {
sub init_globals {
# Defaults:
$Global::version = 20170123;
$Global::version = 20170208;
$Global::progname = 'parallel';
$Global::infinity = 2**31;
$Global::debug = 0;
@ -4046,17 +4131,17 @@ sub tmpname {
do {
$tmpname = $ENV{'TMPDIR'}."/".$name.
join"", map { (0..9,"a".."z","A".."Z")[rand(62)] } (1..5);
} while($Global::unlink{$tmpname}++ or -e $tmpname);
} while(-e $tmpname or $Global::unlink{$tmpname}++);
return $tmpname;
}
#sub tmpfifo {
# # Find an unused name and mkfifo on it
# use POSIX qw(mkfifo);
# my $tmpfifo = tmpname("fif",@_);
# mkfifo($tmpfifo,0600);
# return $tmpfifo;
#}
sub tmpfifo {
# Find an unused name and mkfifo on it
use POSIX qw(mkfifo);
my $tmpfifo = tmpname("fif",@_);
mkfifo($tmpfifo,0600);
return $tmpfifo;
}
sub rm {
# Remove file and remove it from %Global::unlink
@ -5362,7 +5447,7 @@ sub compute_number_of_processes {
sub get_args_or_jobs {
# Get an arg or a job (depending on mode)
if($Global::semaphore or $opt::pipe) {
if($Global::semaphore or ($opt::pipe and not $opt::tee)) {
# Skip: No need to get args
return 1;
} elsif(defined $opt::retries and $count_jobs_already_read) {
@ -6602,9 +6687,7 @@ sub openoutputfiles {
} elsif(not $opt::ungroup) {
# To group we create temporary files for STDOUT and STDERR
# To avoid the cleanup unlink the files immediately (but keep them open)
if(@Global::tee_jobs) {
# files must be removed when the tee is done
} elsif($opt::files) {
if($opt::files) {
($outfhw, $outname) = ::tmpfile(SUFFIX => ".par");
($errfhw, $errname) = ::tmpfile(SUFFIX => ".par");
# --files => only remove stderr
@ -7058,7 +7141,8 @@ sub wrapped {
# * --cat
# * --fifo
# * --sshlogin
# * --pipepart (@Global::cat_partials)
# * --pipepart (@Global::cat_prepends)
# * --tee (@Global::cat_prepends)
# * --pipe
# * --tmux
# The ordering of the wrapping is important:
@ -7070,7 +7154,7 @@ sub wrapped {
# $Global::shell
# $opt::cat
# $opt::fifo
# @Global::cat_partials
# @Global::cat_prepends
# $opt::pipe
# $opt::tmux
# Returns:
@ -7130,15 +7214,21 @@ sub wrapped {
}
# Wrap with ssh + tranferring of files
$command = $self->sshlogin_wrap($command);
if(@Global::cat_partials) {
# Prepend:
if(@Global::cat_prepends) {
# --pipepart: prepend:
# < /tmp/foo perl -e 'while(@ARGV) {
# sysseek(STDIN,shift,0) || die; $left = shift;
# while($read = sysread(STDIN,$buf, ($left > 131072 ? 131072 : $left))){
# $left -= $read; syswrite(STDOUT,$buf);
# }
# }' 0 0 0 11 |
$command = (shift @Global::cat_partials). " | ($command)";
#
# --pipepart --tee: prepend:
# cat dash-a-file |
#
# --pipe --tee: prepend:
# cat fifo |
$command = (shift @Global::cat_prepends). " | ($command)";
} elsif($opt::pipe) {
# Wrap with EOF-detector to avoid starting $command if EOF.
$command = empty_input_wrapper($command);
@ -7862,6 +7952,8 @@ sub start {
$command = "true";
}
$job->openoutputfiles();
# Call slot to store the slot value
$job->slot();
my($stdout_fh,$stderr_fh) = ($job->fh(1,"w"),$job->fh(2,"w"));
if($opt::ungroup or $opt::sqlworker) {
print_dryrun_and_verbose($stdout_fh,$job,$command);
@ -7876,7 +7968,8 @@ sub start {
if($opt::pipe) {
my ($stdin_fh) = ::gensym();
$pid = open3_setpgrp($stdin_fh,$stdout_fh,$stderr_fh,$command);
if($opt::roundrobin) {
if($opt::roundrobin and not $opt::keeporder) {
# --keep-order will make sure the order will be reproducible
::set_fh_non_blocking($stdin_fh);
}
$job->set_fh(0,"w",$stdin_fh);
@ -10207,6 +10300,10 @@ sub total_jobs {
# shorthand for $job->slot();
$job->slot();
}
sub seq {
# shorthand for $job->seq();
$job->seq();
}
sub replace {
# Calculates the corresponding value for a given perl expression

View file

@ -273,11 +273,11 @@ perl quote a string
number of jobs in total
=item Z<> B<$job->>B<slot()>
=item Z<> B<slot()>
slot number of job
=item Z<> B<$job->>B<seq()>
=item Z<> B<seq()>
sequence number of job
@ -988,6 +988,9 @@ to see the difference:
If used with B<--onall> or B<--nonall> the output will grouped by
sshlogin in sorted order.
If used with B<--pipe --roundrobin> and the same input, the jobslots
will get the same blocks in the same order in every run.
=item B<-L> I<max-lines>
@ -1767,13 +1770,13 @@ of GNU B<parallel>'s internal functions and data structures.
Here are a few examples:
Is the job sequence even or odd?
--rpl '{odd} $_ = $job->seq() % 2 ? "odd" : "even"'
--rpl '{odd} $_ = seq() % 2 ? "odd" : "even"'
Pad job sequence with leading zeros to get equal width
--rpl '{0#} $f = "%0".int(1+log(total_jobs())/log(10))."d"; $_=sprintf($f,$job->seq())'
--rpl '{0#} $f=1+int("".(log(total_jobs())/log(10))); $_=sprintf("%0${f}d",seq())'
Job sequence counting from 0
--rpl '{#0} $_ = $job->seq() - 1'
--rpl '{#0} $_ = seq() - 1'
Job slot counting from 2
--rpl '{%1} $_ = $job->slot() + 1'
--rpl '{%1} $_ = slot() + 1'
See also: B<{= perl expression =}> B<--parens>
@ -2165,6 +2168,24 @@ B<{}>.
B<--tagstring> is ignored when using B<-u>, B<--onall>, and B<--nonall>.
=item B<--tee> (alpha testing)
Pipe all data to all jobs. Used with B<--pipe>/B<--pipepart> and
B<:::>.
seq 1000 | parallel --pipe --tee -v wc {} ::: -w -l -c
How many numbers in 1..1000 contain 0..9, and how many bytes do they
fill:
seq 1000 | parallel --pipe --tee --tag 'grep {1} | wc {2}' ::: {0..9} ::: -l -c
How many words contain a..z and how many bytes do they fill?
parallel -a /usr/share/dict/words --pipepart --tee --tag \
'grep {1} | wc {2}' ::: {a..z} ::: -l -c
=item B<--termseq> I<sequence>
Termination sequence. When a job is killed due to B<--timeout>,
@ -4138,7 +4159,7 @@ Killed by Ctrl-C, timeout, not enough memory or similar.
=item Z<>-2 (In joblog and SQL table)
$job->skip() was called in B<{= =}>.
skip() was called in B<{= =}>.
=item Z<>-1000 (In SQL table)

View file

@ -576,7 +576,7 @@ $Global::Initfile && unlink $Global::Initfile;
exit ($err);
sub parse_options {
$Global::version = 20170123;
$Global::version = 20170206;
$Global::progname = 'sql';
# This must be done first as this may exec myself

View file

@ -35,7 +35,7 @@ sql mysql://root:"$mysqlrootpass"@/mysql "DROP DATABASE `whoami`;DROP USER '`who
sql mysql://root:"$mysqlrootpass"@/mysql "CREATE DATABASE `whoami`;CREATE USER '`whoami`'@'localhost' IDENTIFIED BY '`whoami`'; GRANT ALL ON `whoami`.* TO '`whoami`'@'localhost';"
# SHELLS
$INSTALL ash csh fdclone fish fizsh ksh mksh pdksh posh rc rush sash tcsh yash zsh
$INSTALL ash csh dash fdclone fish fizsh ksh mksh pdksh posh rc rush sash tcsh yash zsh
SSHPASS=`goodpasswd`
export SSHPASS

View file

@ -672,6 +672,17 @@ par_retries_replacement_string() {
rm $tmp
}
par_tee() {
export PARALLEL='-k --tee --pipe --tag'
seq 1000000 | parallel 'echo {%};LANG=C wc' ::: {1..5} ::: {a..b}
seq 300000 | parallel 'grep {1} | LANG=C wc {2}' ::: {1..5} ::: -l -c
}
par_tagstring_pipe() {
echo 'bug #50228: --pipe --tagstring broken'
seq 3000 | parallel -j4 --pipe -N1000 -k --tagstring {%} LANG=C wc
}
export -f $(compgen -A function | grep par_)
compgen -A function | grep par_ | sort |
parallel -j6 --tag -k --joblog +/tmp/jl-`basename $0` '{} 2>&1'

View file

@ -239,6 +239,31 @@ par_tmux_fg() {
stdout parallel --tmux --fg sleep ::: 3 | perl -pe 's/.tmp\S+/tmp/'
}
par_pipe_tee() {
echo 'bug #45479: --pipe/--pipepart --tee'
echo '--pipe --tee'
random1G() {
< /dev/zero openssl enc -aes-128-ctr -K 1234 -iv 1234 2>/dev/null |
head -c 1G;
}
random1G | parallel --pipe --tee cat ::: {1..3} | LANG=C wc -c
}
par_pipepart_tee() {
echo 'bug #45479: --pipe/--pipepart --tee'
echo '--pipepart --tee'
random1G() {
< /dev/zero openssl enc -aes-128-ctr -K 1234 -iv 1234 2>/dev/null |
head -c 1G;
}
tmp=$(mktemp)
random1G >$tmp
parallel --pipepart --tee -a $tmp cat ::: {1..3} | LANG=C wc -c
rm $tmp
}
export -f $(compgen -A function | grep par_)
compgen -A function | grep par_ | sort |
parallel --joblog /tmp/jl-`basename $0` -j10 --tag -k '{} 2>&1'

View file

@ -196,6 +196,7 @@ par_result_replace() {
rm -rf /tmp/par_*_49983-*
}
export -f $(compgen -A function | grep par_)
compgen -A function | grep par_ | sort |
parallel -j6 --tag -k --joblog +/tmp/jl-`basename $0` '{} 2>&1'

View file

@ -79,6 +79,31 @@ par_pipepart_block() {
rm /run/shm/parallel$$
}
par_keeporder_roundrobin() {
echo 'bug #50081: --keep-order --round-robin should give predictable results'
export PARALLEL="-j13 --block 1m --pipe --roundrobin"
random500m() {
< /dev/zero openssl enc -aes-128-ctr -K 1234 -iv 1234 2>/dev/null |
head -c 500m;
}
a=$(random500m | parallel -k 'echo {#} $(md5sum)' | sort)
b=$(random500m | parallel -k 'echo {#} $(md5sum)' | sort)
c=$(random500m | parallel 'echo {#} $(md5sum)' | sort)
if [ "$a" == "$b" ] ; then
# Good: -k should be == -k
if [ "$a" == "$c" ] ; then
# Bad: without -k the command should give different output
echo 'Broken: a == c'
printf "$a\n$b\n$c\n"
else
echo OK
fi
else
echo 'Broken: a <> b'
printf "$a\n$b\n$c\n"
fi
}
export -f $(compgen -A function | grep par_)
compgen -A function | grep par_ | sort | parallel -j6 --tag -k '{} 2>&1'

View file

@ -74,6 +74,8 @@ perl -ne '$/="\n\n"; /^Output/../^[^O]\S/ and next; /^ / and print;' ../../src/
# Timings are often off
s/^(\d)$/9/;
s/^(\d\d)$/99/;
# Sometime these vars are not present
s/^PAM_KWALLET5*_LOGIN$//;
# Fails often due to race
s/cat: input_file: No such file or directory\n//;
s{rsync: link_stat ".*/home/parallel/input_file.out" .*\n}{};

View file

@ -1630,3 +1630,37 @@ par_retries_replacement_string 22
par_retries_replacement_string 33
par_retries_replacement_string 33
par_retries_replacement_string 33
par_tagstring_pipe bug #50228: --pipe --tagstring broken
par_tagstring_pipe 1 1000 1000 3893
par_tagstring_pipe 2 1000 1000 5000
par_tagstring_pipe 3 1000 1000 5000
par_tee 1 a 1
par_tee 1 a 1000000 1000000 6888896
par_tee 1 b 2
par_tee 1 b 1000000 1000000 6888896
par_tee 2 a 3
par_tee 2 a 1000000 1000000 6888896
par_tee 2 b 4
par_tee 2 b 1000000 1000000 6888896
par_tee 3 a 5
par_tee 3 a 1000000 1000000 6888896
par_tee 3 b 6
par_tee 3 b 1000000 1000000 6888896
par_tee 4 a 7
par_tee 4 a 1000000 1000000 6888896
par_tee 4 b 8
par_tee 4 b 1000000 1000000 6888896
par_tee 5 a 9
par_tee 5 a 1000000 1000000 6888896
par_tee 5 b 10
par_tee 5 b 1000000 1000000 6888896
par_tee 1 -l 181902
par_tee 1 -c 1228633
par_tee 2 -l 181902
par_tee 2 -c 1228633
par_tee 3 -l 122854
par_tee 3 -c 815297
par_tee 4 -l 122853
par_tee 4 -c 815290
par_tee 5 -l 122853
par_tee 5 -c 815290

View file

@ -492,8 +492,14 @@ par_memleak should give 1 == true
par_memleak Memory use should not depend very much on the total number of jobs run\n
par_memleak Test if memory consumption(300 jobs) < memory consumption(30 jobs) * 110%
par_memleak 1
par_pipe_tee bug #45479: --pipe/--pipepart --tee
par_pipe_tee --pipe --tee
par_pipe_tee 3221225472
par_pipepart_spawn ### bug #46214: Using --pipepart doesnt spawn multiple jobs in version 20150922
par_pipepart_spawn 1:local / 8 / 999
par_pipepart_tee bug #45479: --pipe/--pipepart --tee
par_pipepart_tee --pipepart --tee
par_pipepart_tee 3221225472
par_print_before_halt_on_error ### What is printed before the jobs are killed
par_print_before_halt_on_error -2 exit code 0
par_print_before_halt_on_error -2 0 0

View file

@ -4,6 +4,8 @@ par_children_receive_sig Got TERM
par_children_receive_sig Got TERM
par_children_receive_sig Got INT
par_children_receive_sig Got TERM
par_keeporder_roundrobin bug #50081: --keep-order --round-robin should give predictable results
par_keeporder_roundrobin OK
par_kill_int_twice ### Are children killed if GNU Parallel receives INT twice? There should be no sleep at the end
par_kill_int_twice bash-+-perl---bash---sleep
par_kill_int_twice `-pstree

View file

@ -852,8 +852,6 @@ MFLAGS
MOZ_NO_REMOTE
ORACLE_HOME
ORACLE_SID
PAM_KWALLET5_LOGIN
PAM_KWALLET_LOGIN
PARALLEL
PARALLEL_PID
PARALLEL_SEQ