Merge pull request #50 from zendesk/bevans/limit_ultragrep_concurrency

Allow configuring a sane upper limit on IO threads

lib/ultragrep.rb

@@ -192,6 +192,7 @@ def ultragrep(options)
         exit 1
       end
 
+      concurrency_limit = config.fetch('concurrency_limit', ifnone = file_lists.length)
       request_printer = options.fetch(:printer)
       request_printer.run
 
@@ -201,24 +202,25 @@ def ultragrep(options)
       regexps += options[:not_regexps].map { |r| "!" + r } if options[:not_regexps]
 
       quoted_regexps = quote_shell_words(regexps)
-
       file_lists.each do |files|
         print_search_list(files) if options[:verbose]
 
-        children_pipes = files.map do |file|
-          [worker(file, lua, quoted_regexps, options), file]
-        end
+        files.each_slice(concurrency_limit) do |sliced_files|
+          children_pipes = sliced_files.map do |file|
+              [worker(file, lua, quoted_regexps, options), file]
+          end
 
-        children_pipes.each do |pipe, _|
-          request_printer.set_read_up_to(pipe, 0)
-        end
+          children_pipes.each do |pipe, _|
+            request_printer.set_read_up_to(pipe, 0)
+          end
 
-        # each thread here waits for child data and then pushes it to the printer thread.
-        children_pipes.map do |pipe, filename|
-          worker_reader(filename, pipe, request_printer, options)
-        end.each(&:join)
+          # each thread here waits for child data and then pushes it to the printer thread.
+          children_pipes.map do |pipe, filename|
+            worker_reader(filename, pipe, request_printer, options)
+          end.each(&:join)
 
-        Process.waitall
+          Process.waitall
+        end
       end
 
       request_printer.finish

ultragrep.yml.example

@@ -9,3 +9,4 @@ types:
     format: json
     glob: /Users/*/storage/logs/hosts/*/*/*/*app*/production.log-*.json
 default_type: app
+concurrency_limit: 10