Parallelize diff, download and upload

5 files changed, 131 insertions, 97 deletions

diff --git a/lib/chef/chef_fs/command_line.rb b/lib/chef/chef_fs/command_line.rb
index 6236871a52..dcb7569d4e 100644
--- a/lib/chef/chef_fs/command_line.rb
+++ b/lib/chef/chef_fs/command_line.rb
@@ -31,7 +31,7 @@ class Chef
 
         get_content = (output_mode != :name_only && output_mode != :name_status)
         found_match = false
-        diff(pattern, a_root, b_root, recurse_depth, get_content) do |type, old_entry, new_entry, old_value, new_value, error|
+        diff(pattern, a_root, b_root, recurse_depth, get_content).each do |type, old_entry, new_entry, old_value, new_value, error|
           found_match = true unless type == :both_nonexistent
           old_path = format_path.call(old_entry)
           new_path = format_path.call(new_entry)
@@ -125,112 +125,128 @@ class Chef
         error
       end
 
-      def self.diff(pattern, a_root, b_root, recurse_depth, get_content)
-        Chef::ChefFS::FileSystem.list_pairs(pattern, a_root, b_root).each do |a, b|
-          diff_entries(a, b, recurse_depth, get_content) do |diff|
-            yield diff
-          end
+      def self.diff(pattern, old_root, new_root, recurse_depth, get_content)
+        Chef::ChefFS::Parallelizer.parallelize(Chef::ChefFS::FileSystem.list_pairs(pattern, old_root, new_root), :flatten => true) do |old_entry, new_entry|
+          diff_entries(old_entry, new_entry, recurse_depth, get_content)
         end
       end
 
       # Diff two known entries (could be files or dirs)
       def self.diff_entries(old_entry, new_entry, recurse_depth, get_content)
-        begin
-          # If both are directories
-          if old_entry.dir?
-            if new_entry.dir?
-              if recurse_depth == 0
-                yield [ :common_subdirectories, old_entry, new_entry ]
-              else
-                Chef::ChefFS::FileSystem.child_pairs(old_entry, new_entry).each do |old_child,new_child|
-                  diff_entries(old_child, new_child,
-                               recurse_depth ? recurse_depth - 1 : nil, get_content) do |diff|
-                    yield diff
-                  end
-                end
+        # If both are directories
+        if old_entry.dir?
+          if new_entry.dir?
+            if recurse_depth == 0
+              return [ [ :common_subdirectories, old_entry, new_entry ] ]
+            else
+              return Chef::ChefFS::Parallelizer.parallelize(Chef::ChefFS::FileSystem.child_pairs(old_entry, new_entry), :flatten => true) do |old_child, new_child|
+                Chef::ChefFS::CommandLine.diff_entries(old_child, new_child, recurse_depth ? recurse_depth - 1 : nil, get_content)
               end
+            end
 
           # If old is a directory and new is a file
-            elsif new_entry.exists?
-              yield [ :directory_to_file, old_entry, new_entry ]
+          elsif new_entry.exists?
+            return [ [ :directory_to_file, old_entry, new_entry ] ]
 
           # If old is a directory and new does not exist
-            elsif new_entry.parent.can_have_child?(old_entry.name, old_entry.dir?)
-              yield [ :deleted, old_entry, new_entry ]
-            end
+          elsif new_entry.parent.can_have_child?(old_entry.name, old_entry.dir?)
+            return [ [ :deleted, old_entry, new_entry ] ]
 
-          # If new is a directory and old is a file
-          elsif new_entry.dir?
-            if old_entry.exists?
-              yield [ :file_to_directory, old_entry, new_entry ]
+          # If the new entry does not and *cannot* exist, report that.
+          else
+            return [ [ :new_cannot_upload, old_entry, new_entry ] ]
+          end
+
+        # If new is a directory and old is a file
+        elsif new_entry.dir?
+          if old_entry.exists?
+            return [ [ :file_to_directory, old_entry, new_entry ] ]
 
           # If new is a directory and old does not exist
-            elsif old_entry.parent.can_have_child?(new_entry.name, new_entry.dir?)
-              yield [ :added, old_entry, new_entry ]
-            end
+          elsif old_entry.parent.can_have_child?(new_entry.name, new_entry.dir?)
+            return [ [ :added, old_entry, new_entry ] ]
 
-          # Neither is a directory, so they are diffable with file diff
+          # If the new entry does not and *cannot* exist, report that.
           else
-            are_same, old_value, new_value = Chef::ChefFS::FileSystem.compare(old_entry, new_entry)
-            if are_same
-              if old_value == :none
-                yield [ :both_nonexistent, old_entry, new_entry ]
-              else
-                yield [ :same, old_entry, new_entry ]
-              end
+            return [ [ :old_cannot_upload, old_entry, new_entry ] ]
+          end
+
+        # Neither is a directory, so they are diffable with file diff
+        else
+          are_same, old_value, new_value = Chef::ChefFS::FileSystem.compare(old_entry, new_entry)
+          if are_same
+            if old_value == :none
+              return [ [ :both_nonexistent, old_entry, new_entry ] ]
             else
-              if old_value == :none
-                old_exists = false
-              elsif old_value.nil?
-                old_exists = old_entry.exists?
-              else
-                old_exists = true
-              end
+              return [ [ :same, old_entry, new_entry ] ]
+            end
+          else
+            if old_value == :none
+              old_exists = false
+            elsif old_value.nil?
+              old_exists = old_entry.exists?
+            else
+              old_exists = true
+            end
 
-              if new_value == :none
-                new_exists = false
-              elsif new_value.nil?
-                new_exists = new_entry.exists?
-              else
-                new_exists = true
-              end
+            if new_value == :none
+              new_exists = false
+            elsif new_value.nil?
+              new_exists = new_entry.exists?
+            else
+              new_exists = true
+            end
 
-              # If one of the files doesn't exist, we only want to print the diff if the
-              # other file *could be uploaded/downloaded*.
-              if !old_exists && !old_entry.parent.can_have_child?(new_entry.name, new_entry.dir?)
-                yield [ :old_cannot_upload, old_entry, new_entry ]
-                return
-              end
-              if !new_exists && !new_entry.parent.can_have_child?(old_entry.name, old_entry.dir?)
-                yield [ :new_cannot_upload, old_entry, new_entry ]
-                return
-              end
+            # If one of the files doesn't exist, we only want to print the diff if the
+            # other file *could be uploaded/downloaded*.
+            if !old_exists && !old_entry.parent.can_have_child?(new_entry.name, new_entry.dir?)
+              return [ [ :old_cannot_upload, old_entry, new_entry ] ]
+            end
+            if !new_exists && !new_entry.parent.can_have_child?(old_entry.name, old_entry.dir?)
+              return [ [ :new_cannot_upload, old_entry, new_entry ] ]
+            end
 
-              if get_content
-                # If we haven't read the values yet, get them now so that they can be diffed
-                begin
-                  old_value = old_entry.read if old_value.nil?
-                rescue Chef::ChefFS::FileSystem::NotFoundError
-                  old_value = :none
-                end
-                begin
-                  new_value = new_entry.read if new_value.nil?
-                rescue Chef::ChefFS::FileSystem::NotFoundError
-                  new_value = :none
-                end
+            if get_content
+              # If we haven't read the values yet, get them now so that they can be diffed
+              begin
+                old_value = old_entry.read if old_value.nil?
+              rescue Chef::ChefFS::FileSystem::NotFoundError
+                old_value = :none
               end
-
-              if old_value == :none || (old_value == nil && !old_entry.exists?)
-                yield [ :added, old_entry, new_entry, old_value, new_value ]
-              elsif new_value == :none
-                yield [ :deleted, old_entry, new_entry, old_value, new_value ]
-              else
-                yield [ :modified, old_entry, new_entry, old_value, new_value ]
+              begin
+                new_value = new_entry.read if new_value.nil?
+              rescue Chef::ChefFS::FileSystem::NotFoundError
+                new_value = :none
               end
             end
+
+            if old_value == :none || (old_value == nil && !old_entry.exists?)
+              return [ [ :added, old_entry, new_entry, old_value, new_value ] ]
+            elsif new_value == :none
+              return [ [ :deleted, old_entry, new_entry, old_value, new_value ] ]
+            else
+              return [ [ :modified, old_entry, new_entry, old_value, new_value ] ]
+            end
           end
-        rescue Chef::ChefFS::FileSystem::FileSystemError => e
-          yield [ :error, old_entry, new_entry, nil, nil, e ]
+        end
+      rescue Chef::ChefFS::FileSystem::FileSystemError => e
+        return [ [ :error, old_entry, new_entry, nil, nil, e ] ]
+      end
+
+      class Differ
+        def initialize(old_entry, new_entry, recurse_depth, get_content)
+          @old_entry = old_entry
+          @new_entry = new_entry
+          @recurse_depth = recurse_depth
+          @get_content = get_content
+        end
+
+        attr_reader :old_entry
+        attr_reader :new_entry
+        attr_reader :recurse_depth
+        attr_reader :get_content
+
+        def each
         end
       end
 
diff --git a/lib/chef/chef_fs/file_system.rb b/lib/chef/chef_fs/file_system.rb
index 523fe8b38e..385f955806 100644
--- a/lib/chef/chef_fs/file_system.rb
+++ b/lib/chef/chef_fs/file_system.rb
@@ -139,7 +139,7 @@ class Chef
       def self.copy_to(pattern, src_root, dest_root, recurse_depth, options, ui, format_path)
         found_result = false
         error = false
-        list_pairs(pattern, src_root, dest_root).each do |src, dest|
+        parallel_do(list_pairs(pattern, src_root, dest_root)) do |src, dest|
           found_result = true
           new_dest_parent = get_or_create_parent(dest, options, ui, format_path)
           child_error = copy_entries(src, dest, new_dest_parent, recurse_depth, options, ui, format_path)
@@ -243,6 +243,7 @@ class Chef
           are_same, b_value, a_value = b.compare_to(a)
         end
         if are_same.nil?
+          # TODO these reads can be parallelized
           begin
             a_value = a.read if a_value.nil?
           rescue Chef::ChefFS::FileSystem::NotFoundError
@@ -315,7 +316,7 @@ class Chef
                 end
                 # Directory creation is recursive.
                 if recurse_depth != 0
-                  src_entry.children.each do |src_child|
+                  parallel_do(src_entry.children) do |src_child|
                     new_dest_child = new_dest_dir.child(src_child.name)
                     child_error = copy_entries(src_child, new_dest_child, new_dest_dir, recurse_depth ? recurse_depth - 1 : recurse_depth, options, ui, format_path)
                     error ||= child_error
@@ -352,7 +353,7 @@ class Chef
               if dest_entry.dir?
                 # If both are directories, recurse into their children
                 if recurse_depth != 0
-                  child_pairs(src_entry, dest_entry).each do |src_child, dest_child|
+                  parallel_do(child_pairs(src_entry, dest_entry)) do |src_child, dest_child|
                     child_error = copy_entries(src_child, dest_child, dest_entry, recurse_depth ? recurse_depth - 1 : recurse_depth, options, ui, format_path)
                     error ||= child_error
                   end
@@ -415,6 +416,9 @@ class Chef
         return parent
       end
 
+      def self.parallel_do(enum, options = {}, &block)
+        Chef::ChefFS::Parallelizer.parallelize(enum, options, &block).to_a
+      end
     end
   end
 end
diff --git a/lib/chef/chef_fs/file_system/cookbook_dir.rb b/lib/chef/chef_fs/file_system/cookbook_dir.rb
index cfdabc4a61..debce979f8 100644
--- a/lib/chef/chef_fs/file_system/cookbook_dir.rb
+++ b/lib/chef/chef_fs/file_system/cookbook_dir.rb
@@ -151,7 +151,7 @@ class Chef
             return [ !exists?, nil, nil ]
           end
           are_same = true
-          Chef::ChefFS::CommandLine::diff_entries(self, other, nil, :name_only) do |type, old_entry, new_entry|
+          Chef::ChefFS::CommandLine::diff_entries(self, other, nil, :name_only).each do |type, old_entry, new_entry|
             if [ :directory_to_file, :file_to_directory, :deleted, :added, :modified ].include?(type)
               are_same = false
             end
diff --git a/lib/chef/chef_fs/file_system/rest_list_entry.rb b/lib/chef/chef_fs/file_system/rest_list_entry.rb
index cd1e439c47..fb7a3f5c4d 100644
--- a/lib/chef/chef_fs/file_system/rest_list_entry.rb
+++ b/lib/chef/chef_fs/file_system/rest_list_entry.rb
@@ -106,6 +106,8 @@ class Chef
         end
 
         def compare_to(other)
+          # TODO this pair of reads can be parallelized
+
           # Grab the other value
           begin
             other_value_json = other.read
diff --git a/lib/chef/chef_fs/parallelizer.rb b/lib/chef/chef_fs/parallelizer.rb
index 6aab14d7f5..e0e7a77322 100644
--- a/lib/chef/chef_fs/parallelizer.rb
+++ b/lib/chef/chef_fs/parallelizer.rb
@@ -2,10 +2,13 @@ class Chef
   module ChefFS
     class Parallelizer
       @@parallelizer = nil
+      @@threads = 0
 
       def self.threads=(value)
-        raise "Cannot set threads after parallelize has been called" if @@parallelizer
-        @@threads = value
+        if @@threads != value
+          @@threads = value
+          @@parallelizer = nil
+        end
       end
 
       def self.parallelize(enumerator, options = {}, &block)
@@ -47,13 +50,17 @@ class Chef
           next_index = 0
           while true
             # Report any results that already exist
-            while @status.length > next_index && @status[next_index] == :finished
-              if @options[:flatten]
-                @outputs[next_index].each do |entry|
-                  yield entry
+            while @status.length > next_index && ([:finished, :exception].include?(@status[next_index]))
+              if @status[next_index] == :finished
+                if @options[:flatten]
+                  @outputs[next_index].each do |entry|
+                    yield entry
+                  end
+                else
+                  yield @outputs[next_index]
                 end
               else
-                yield @outputs[next_index]
+                raise @outputs[next_index]
               end
               next_index = next_index + 1
             end
@@ -84,8 +91,13 @@ class Chef
             [ index, input ]
           end
 
-          @outputs[index] = @block.call(input, @options)
-          @status[index] = :finished
+          begin
+            @outputs[index] = @block.call(input)
+            @status[index] = :finished
+          rescue
+            @outputs[index] = $!
+            @status[index] = :exception
+          end
           index
         end
       end