Parallel automake: ordered output messages. [2/4]

[Ralf Wildenhues]

This serializes the regular warning and error output.
The QUEUE_MESSAGE constant will be amended with further
constants later.

Cheers,
Ralf

        Parallel automake: ordered output messages.
        * lib/Automake/Channels.pm (%_default_options): New options
        `ordered' default enabled, `queue', default zero (no queue),
        `queue_key' default undefined.
        (_merge_options): Ensure `ordered' channels cannot have fatal
        messages or backtrace output.
        (_enqueue, _dequeue): New functions, to push messages onto a
        Thread::Queue, and output messages from such a queue, suitably
        weeded for duplicates in the same manner as _print_message.
        (_print_message): If the channel is ordered and has an
        associated queue, then enqueue messages instead of printing
        them.
        (setup_channel_queue, pop_channel_queue): New functions,
        to set a Thread::Queue for channels, and to flush a queue.
        * lib/Automake/ChannelDefs.pm: Unset channel option `ordered'
        for fatal, automake, and verb channels.
        * automake.in (QUEUE_MESSAGE): New global constant, used as
        serialization key.
        (handle_makefiles_threaded): Create message queues for each
        input file; workers queue messages, and the master outputs them
        ordered, using the new Channels.pm functions.
        * tests/parallel-am.test: Also check for ordered output (in the
        absence of --add-missing races).
        * tests/parallel-am2.test: New test; check for ordered output of
        warning and (regular) error messages.
        * tests/Makefile.am: Adjust.

diff --git a/automake.in b/automake.in
index c9b1c63..d7db627 100755
--- a/automake.in
+++ b/automake.in
@@ -276,6 +276,11 @@ use constant COMPILE_ORDINARY => 2;
 # We can't always associate a location to a variable or a rule,
 # when it's defined by Automake.  We use INTERNAL in this case.
 use constant INTERNAL => new Automake::Location;
+
+# Serialization keys for message queues.
+use constant {
+  QUEUE_MESSAGE   => "msg",
+};
 
 
 ## ---------------------------------- ##
@@ -8057,7 +8062,14 @@ sub get_number_of_threads
 # handle_makefiles_threaded ($NTHREADS)
 # -------------------------------------
 # Deal with all makefiles, using threads.  The general strategy is to
-# spawn NTHREADS worker threads, and dispatch makefiles to them.
+# spawn NTHREADS worker threads, dispatch makefiles to them, and let the
+# worker threads push back everything that needs serialization:
+# * warning and (normal) error messages, for stable stderr output
+#   order and content (avoiding duplicates, for example),
+# * races when collecting aux files for distribution.
+#
+# The latter requires that the makefile that deals with the aux dir
+# files be handled last, done by the master thread.
 sub handle_makefiles_threaded ($)
 {
   my ($nthreads) = @_;
@@ -8069,8 +8081,14 @@ sub handle_makefiles_threaded ($)
       $last_input_file = pop @queued_input_files;
     }
 
-  # The file queue distributes all makefiles.
+  # The file queue distributes all makefiles, the message queues
+  # collect all serializations needed for respective files.
   my $file_queue = Thread::Queue->new;
+  my %msg_queues;
+  foreach my $file (@queued_input_files)
+    {
+      $msg_queues{$file} = Thread::Queue->new;
+    }
 
   verb "spawning $nthreads worker threads";
   my @threads = (1 .. $nthreads);
@@ -8081,7 +8099,11 @@ sub handle_makefiles_threaded ($)
          while (my $file = $file_queue->dequeue)
            {
              verb "handling $file";
+             my $queue = $msg_queues{$file};
+             setup_channel_queue ($queue, QUEUE_MESSAGE);
              handle_makefile ($file);
+             $queue->enqueue (undef);
+             setup_channel_queue (undef, undef);
            }
          return $exit_code;
        });
@@ -8091,6 +8113,25 @@ sub handle_makefiles_threaded ($)
   verb "queuing " . @queued_input_files . " input files";
   $file_queue->enqueue (@queued_input_files, (undef) x @threads);
 
+  # Collect and process serializations.
+  foreach my $file (@queued_input_files)
+    {
+      verb "dequeuing messages for " . $file;
+      reset_local_duplicates ();
+      my $queue = $msg_queues{$file};
+      while (my $key = $queue->dequeue)
+       {
+         if ($key eq QUEUE_MESSAGE)
+           {
+             pop_channel_queue ($queue);
+           }
+         else
+           {
+             prog_error "unexpected key $key";
+           }
+       }
+    }
+
   foreach my $t (@threads)
     {
       my @exit_thread = $t->join;
diff --git a/lib/Automake/ChannelDefs.pm b/lib/Automake/ChannelDefs.pm
index 17038a6..afd701e 100644
--- a/lib/Automake/ChannelDefs.pm
+++ b/lib/Automake/ChannelDefs.pm
@@ -134,7 +134,7 @@ Informative messages.
 # Do not forget to update &usage and the manual
 # if you add or change a warning channel.
 
-register_channel 'fatal', type => 'fatal', uniq_part => UP_NONE;
+register_channel 'fatal', type => 'fatal', uniq_part => UP_NONE, ordered => 0;
 register_channel 'error', type => 'error';
 register_channel 'error-gnu', type => 'error';
 register_channel 'error-gnu/warn', type => 'error';
@@ -144,7 +144,7 @@ register_channel 'automake', type => 'fatal', backtrace => 
1,
             "## Internal Error ##\n" .
             "####################\n"),
   footer => "\nPlease contact <address@hidden>.",
-  uniq_part => UP_NONE;
+  uniq_part => UP_NONE, ordered => 0;
 
 register_channel 'gnu', type => 'warning';
 register_channel 'obsolete', type => 'warning', silent => 1;
@@ -153,7 +153,8 @@ register_channel 'portability', type => 'warning', silent 
=> 1;
 register_channel 'syntax', type => 'warning';
 register_channel 'unsupported', type => 'warning';
 
-register_channel 'verb', type => 'debug', silent => 1, uniq_part => UP_NONE;
+register_channel 'verb', type => 'debug', silent => 1, uniq_part => UP_NONE,
+  ordered => 0;
 register_channel 'note', type => 'debug', silent => 0;
 
 =head2 FUNCTIONS
diff --git a/lib/Automake/Channels.pm b/lib/Automake/Channels.pm
index d12bb8d..6b79b4f 100644
--- a/lib/Automake/Channels.pm
+++ b/lib/Automake/Channels.pm
@@ -43,6 +43,13 @@ Automake::Channels - support functions for error and warning 
management
   # Turn on all channels of type 'warning'.
   setup_channel_type 'warning', silent => 0;
 
+  # Redirect all channels to push messages on a Thread::Queue using
+  # the specified serialization key.
+  setup_channel_queue $queue, $key;
+
+  # Output a message pending in a Thread::Queue.
+  pop_channel_queue $queue;
+
   # Treat all warnings as errors.
   $warnings_are_errors = 1;
 
@@ -74,6 +81,7 @@ use vars qw (@ISA @EXPORT %channels $me);
              &setup_channel &setup_channel_type
              &dup_channel_setup &drop_channel_setup
              &buffer_messages &flush_messages
+             &setup_channel_queue &pop_channel_queue
              US_GLOBAL US_LOCAL
              UP_NONE UP_TEXT UP_LOC_TEXT);
 
@@ -173,6 +181,11 @@ The file where the error should be output.
 Whether the channel should be silent.  Use this do disable a
 category of warning, for instance.
 
+=item C<ordered =E<gt> 1>
+
+Whether, with multi-threaded execution, the message should be queued
+for ordered output.
+
 =item C<uniq_part =E<gt> UP_LOC_TEXT>
 
 The part of the message subject to duplicate filtering.  See the
@@ -252,6 +265,9 @@ use vars qw (%_default_options %_global_duplicate_messages
    exit_code => 1,
    file => \*STDERR,
    silent => 0,
+   ordered => 1,
+   queue => 0,
+   queue_key => undef,
    uniq_scope => US_LOCAL,
    uniq_part => UP_LOC_TEXT,
    header => '',
@@ -323,6 +339,13 @@ sub _merge_options (\%%)
          confess "unknown option `$_'";
        }
     }
+  if ($hash->{'ordered'})
+    {
+      confess "fatal messages cannot be ordered"
+       if $hash->{'type'} eq 'fatal';
+      confess "backtrace cannot be output on ordered messages"
+       if $hash->{'backtrace'};
+    }
 }
 
 =item C<register_channel ($name, [%options])>
@@ -403,6 +426,63 @@ sub _format_message ($$%)
   return $msg;
 }
 
+# _enqueue ($QUEUE, $KEY, $UNIQ_SCOPE, $TO_FILTER, $MSG, $FILE)
+# ------------------------------------------------------------
+# Push message on a queue, to be processed by another thread.
+sub _enqueue ($$$$$$)
+{
+  my ($queue, $key, $uniq_scope, $to_filter, $msg, $file) = @_;
+  $queue->enqueue ($key, $msg, $to_filter, $uniq_scope);
+  confess "message queuing works only for STDERR"
+    if $file ne \*STDERR;
+}
+
+# _dequeue ($QUEUE)
+# -----------------
+# Pop a message from a queue, and print, similarly to how
+# _print_message would do it.  Return 0 if the queue is
+# empty.  Note that the key has already been dequeued.
+sub _dequeue ($)
+{
+  my ($queue) = @_;
+  my $msg = $queue->dequeue || return 0;
+  my $to_filter = $queue->dequeue;
+  my $uniq_scope = $queue->dequeue;
+  my $file = \*STDERR;
+
+  if ($to_filter ne '')
+    {
+      # Do we want local or global uniqueness?
+      my $dups;
+      if ($uniq_scope == US_LOCAL)
+       {
+         $dups = \%_local_duplicate_messages;
+       }
+      elsif ($uniq_scope == US_GLOBAL)
+       {
+         $dups = \%_global_duplicate_messages;
+       }
+      else
+       {
+         confess "unknown value for uniq_scope: " . $uniq_scope;
+       }
+
+      # Update the hash of messages.
+      if (exists $dups->{$to_filter})
+       {
+         ++$dups->{$to_filter};
+         return 1;
+       }
+      else
+       {
+         $dups->{$to_filter} = 0;
+       }
+    }
+  print $file $msg;
+  return 1;
+}
+
+
 # Store partial messages here. (See the 'partial' option.)
 use vars qw ($partial);
 $partial = '';
@@ -431,10 +511,10 @@ sub _print_message ($$%)
     }
 
   # Check for duplicate message if requested.
+  my $to_filter;
   if ($opts{'uniq_part'} ne UP_NONE)
     {
       # Which part of the error should we match?
-      my $to_filter;
       if ($opts{'uniq_part'} eq UP_TEXT)
        {
          $to_filter = $message;
@@ -475,7 +555,15 @@ sub _print_message ($$%)
        }
     }
   my $file = $opts{'file'};
-  print $file $msg;
+  if ($opts{'ordered'} && $opts{'queue'})
+    {
+      _enqueue ($opts{'queue'}, $opts{'queue_key'}, $opts{'uniq_scope'},
+               $to_filter, $msg, $file);
+    }
+  else
+    {
+      print $file $msg;
+    }
   return 1;
 }
 
@@ -680,6 +768,33 @@ sub flush_messages ()
   @backlog = ();
 }
 
+=item C<setup_channel_queue ($queue, $key)>
+
+Set the queue to fill for each channel that is ordered,
+and the key to use for serialization.
+
+=cut
+sub setup_channel_queue ($$)
+{
+  my ($queue, $key) = @_;
+  foreach my $channel (keys %channels)
+    {
+      setup_channel $channel, queue => $queue, queue_key => $key
+        if $channels{$channel}{'ordered'};
+    }
+}
+
+=item C<pop_channel_queue ($queue)>
+
+pop a message off the $queue; the key has already been popped.
+
+=cut
+sub pop_channel_queue ($)
+{
+  my ($queue) = @_;
+  return _dequeue ($queue);
+}
+
 =back
 
 =head1 SEE ALSO
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 483849c..7c6d4a2 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -450,6 +450,7 @@ output13.test \
 output-order.test \
 overrid.test \
 parallel-am.test \
+parallel-am2.test \
 parse.test \
 percent.test \
 percent2.test \
diff --git a/tests/parallel-am.test b/tests/parallel-am.test
index f6e4743..d57e014 100755
--- a/tests/parallel-am.test
+++ b/tests/parallel-am.test
@@ -33,7 +33,7 @@
 # 5) fatal error and debug messages could be identical.  This is not
 #    intended, though.
 #
-# This test checks (0).
+# This test checks (0), (1), and (2).  See sister tests for further coverage.
 
 . ./defs || Exit 1
 
@@ -71,9 +71,17 @@ mkdir build-aux
 
 $ACLOCAL
 
+# This test may have to be run several times in order to expose the
+# race that, when the last Makefile.in (the toplevel one) is created
+# before the other ones have finished, not all auxiliary files may
+# be installed yet, thus some may not be distributed.
+#
+# Further, automake output should be stable.
+
 # Generate expected output using the non-threaded code.
 unset AUTOMAKE_JOBS
 AUTOMAKE_run 0 --add-missing
+mv stderr expected
 makefile_ins=`find . -name Makefile.in`
 for file in $makefile_ins; do
   mv $file $file.exp
@@ -83,7 +91,9 @@ AUTOMAKE_JOBS=5
 export AUTOMAKE_JOBS
 
 for run in 1 2 3 4 5 6 7; do
-  AUTOMAKE_run 0
+  rm -f build-aux/* sub*/Makefile.in
+  AUTOMAKE_run 0 --add-missing
+  diff stderr expected
   for file in $makefile_ins; do
     diff $file $file.exp
   done
diff --git a/tests/parallel-am2.test b/tests/parallel-am2.test
new file mode 100755
index 0000000..b5a4ac7
--- /dev/null
+++ b/tests/parallel-am2.test
@@ -0,0 +1,77 @@
+#! /bin/sh
+# Copyright (C) 2008  Free Software Foundation, Inc.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3, or (at your option)
+# any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+# Test parallel automake execution.
+
+# This tests:
+# 4) warning and normal error output should be identical, in that duplicate
+#    warnings should be omitted in the same way as without threads,
+
+. ./defs || Exit 1
+
+set -e
+
+mkdir sub
+
+cat > Makefile.am << 'END'
+AUTOMAKE_OPTIONS = subdir-objects
+bin_PROGRAMS = main
+main_SOURCES = sub/main.c
+SUBDIRS =
+END
+
+list='1 2 3'
+for i in $list; do
+  echo "AC_CONFIG_FILES([sub$i/Makefile])" >> configure.in
+  echo "SUBDIRS += sub$i" >> Makefile.am
+  mkdir sub$i sub$i/sub
+  cat > sub$i/Makefile.am << END
+AUTOMAKE_OPTIONS = subdir-objects
+bin_PROGRAMS = sub$i
+sub${i}_SOURCES = sub/main$i.c
+END
+done
+
+mkdir build-aux
+
+$ACLOCAL
+
+# Independently of the number of worker threads, automake output
+# should be
+# - stable (multiple runs should produce the same output),
+# - properly uniquified,
+# - complete (output from worker threads should not be lost).
+#
+# The parts output by --add-missing are unstable not only wrt. order
+# but also wrt. content: any of the Makefile.am files may cause the
+# depcomp script to be installed (or several of them).
+# Thus we install the auxiliary files in a prior step.
+
+# Generate expected output using non-threaded code.
+unset AUTOMAKE_JOBS
+AUTOMAKE_fails --add-missing
+AUTOMAKE_fails
+mv stderr expected
+
+AUTOMAKE_JOBS=5
+export AUTOMAKE_JOBS
+
+for i in 1 2 3 4 5 6 7 8; do
+  AUTOMAKE_fails
+  diff expected stderr
+done
+
+: