<commit_before>namespace :enju_trunk do
  namespace :reindex do
    desc "Reindex all model (by model)" 
    task :parallelbymodel, [:exec_processor_size] => :environment do |t, args|
      Sunspot.session = Sunspot::SessionProxy::Retry5xxSessionProxy.new(Sunspot.session)
      reindex_options = { :batch_commit => false, :exec_processor_size => 1 }

      require "parallel"
      require "benchmark"

      case args[:exec_processor_size]
      when 'false'
        reindex_options[:exec_processor_size] = 1
      when /^\d+$/ 
        reindex_options[:exec_processor_size] = args[:exec_processor_size].to_i if args[:exec_processor_size].to_i > 0
      end

      puts "#{Parallel.processor_count} procesor(s)"
      puts "reindex using #{reindex_options[:exec_processor_size]} procesor(s)"

      Dir.glob(Rails.root.join('app/models/**/*.rb')).each { |path| require path }
      sunspot_models = Sunspot.searchable
      total_documents = sunspot_models.map { | m | m.count }.sum
      sorted_models = sunspot_models.sort{|a, b| a.count <=> b.count} 

      # Set up progress_bar to, ah, report progress
      begin
        require 'progress_bar'
        reindex_options[:progress_bar] = ProgressBar.new(total_documents)
      rescue LoadError => e
        $stdout.puts "Skipping progress bar: for progress reporting, add gem 'progress_bar' to your Gemfile"
      rescue Exception => e
        $stderr.puts "Error using progress bar: #{e.message}"
      end

      $stdout.puts "Reindex model list:"
      sorted_models.each { |model|
        $stdout.puts " name=#{model.name} record_size=#{model.count}"
      }
      $stdout.flush

      Benchmark.bm do |x|
        x.report('thread') {
          results = Parallel.map(sorted_models, :in_threads => reindex_options[:exec_processor_size]) { |model|
            #$stdout.puts "#{model.name} #{model.class}"
            #$stdout.puts "name=#{model.name} size=#{model.count}"
          
            model.solr_reindex(reindex_options)
          }
        }
      end
    end

    desc "Reindex a manifestations"
    task :manifestations => :environment do
      request_reindex Manifestation 
    end

    desc "Reindex a items"
    task :items => :environment do
      request_reindex Item
    end<commit_msg>refs 4415: Solrのreindexをrecord毎に並列に行う - rake task を追加した   enju_trunk:reindex:parallelbyrecord[exec_processor_size,batch_size]   - default の batch_size は 1000   - batch_size 分のレコード単位で分割並列化を行う - 並列度を一定に抑えるため record でのみ並列化を行った<commit_after># based on Sunspot::Rails::Searchable.solr_index
# (sunspot_rails-2.0.0.pre.130115/lib/sunspot/rails/searchable.rb)
module Sunspot
  module Rails
    module Searchable
      module ActsAsMethods									
	def solr_index_parallel(opts={})
          require 'parallel'
          require 'benchmark'
          
          options = {
            :batch_size => Sunspot.config.indexing.default_batch_size,
            :batch_commit => true,
            :include => self.sunspot_options[:include],
            :start => opts.delete(:first_id) || 0
          }.merge(opts)
          
          find_in_batch_options = {
            :include => options[:include],
            :batch_size => options[:batch_size],
            :start => options[:start]
          }
          
          exec_processor_size = options[:exec_processor_size]
          progress_bar = options[:progress_bar]
          
          mutex = Mutex.new
          tcount = exec_processor_size
	  
          if options[:batch_size]
            batch_counter = 0
            ths = []
            find_in_batches(find_in_batch_options) do |records|
              while tcount < 1 do
                sleep 5
              end
	      mutex.synchronize {tcount -= 1}
              solr_benchmark options[:batch_size], batch_counter do
                ths << Thread.start do
                  #puts "Thread #{tcount} of #{exec_processor_size}"
                  Sunspot.index(records.select { |r| r.indexable? })
                  Sunspot.commit if options[:batch_commit]
                  mutex.synchronize {tcount += 1}
                  Thread.exit
                end
                # track progress
                progress_bar.increment!(records.length) if progress_bar
                batch_counter += 1
              end
            end
            ths.each {|t| t.join}
          else
            records = all(:include => options[:include]).select { |model| model.indexable? }
            Sunspot.index!(records)
          end
          # perform a final commit if not committing in batches
          Sunspot.commit unless options[:batch_commit]
        end
      end
    end
  end
end

namespace :enju_trunk do
  namespace :reindex do
    desc "Reindex all model (by model)" 
    task :parallelbymodel, [:exec_processor_size] => :environment do |t, args|
      Sunspot.session = Sunspot::SessionProxy::Retry5xxSessionProxy.new(Sunspot.session)
      reindex_options = { :batch_commit => false, :exec_processor_size => 1 }

      require "parallel"
      require "benchmark"

      case args[:exec_processor_size]
      when 'false'
        reindex_options[:exec_processor_size] = 1
      when /^\d+$/ 
        reindex_options[:exec_processor_size] = args[:exec_processor_size].to_i if args[:exec_processor_size].to_i > 0
      end

      puts "#{Parallel.processor_count} procesor(s)"
      puts "reindex using #{reindex_options[:exec_processor_size]} procesor(s)"

      Dir.glob(Rails.root.join('app/models/**/*.rb')).each { |path| require path }
      sunspot_models = Sunspot.searchable
      total_documents = sunspot_models.map { | m | m.count }.sum
      sorted_models = sunspot_models.sort{|a, b| a.count <=> b.count} 

      # Set up progress_bar to, ah, report progress
      begin
        require 'progress_bar'
        reindex_options[:progress_bar] = ProgressBar.new(total_documents)
      rescue LoadError => e
        $stdout.puts "Skipping progress bar: for progress reporting, add gem 'progress_bar' to your Gemfile"
      rescue Exception => e
        $stderr.puts "Error using progress bar: #{e.message}"
      end

      $stdout.puts "Reindex model list:"
      sorted_models.each { |model|
        $stdout.puts " name=#{model.name} record_size=#{model.count}"
      }
      $stdout.flush

      Benchmark.bm do |x|
        x.report('thread') {
          results = Parallel.map(sorted_models, :in_threads => reindex_options[:exec_processor_size]) { |model|
            #$stdout.puts "#{model.name} #{model.class}"
            #$stdout.puts "name=#{model.name} size=#{model.count}"
          
            model.solr_reindex(reindex_options)
          }
        }
      end
    end

    desc "Reindex all model (paralleled by record)" 
    task :parallelbyrecord, [:exec_processor_size,:batch_size] => :environment do |t, args|
      Sunspot.session = Sunspot::SessionProxy::Retry5xxSessionProxy.new(Sunspot.session)
      reindex_options = { :batch_commit => false,
                          :exec_processor_size => 1, 
			  :batch_size => 1000 }

      case args[:exec_processor_size]
      when 'false'
        reindex_options[:exec_processor_size] = 1
      when /^\d+$/ 
        reindex_options[:exec_processor_size] = args[:exec_processor_size].to_i if args[:exec_processor_size].to_i > 0
      end

      case args[:batch_size]
      when 'false'
        reindex_options[:batch_size] = 1000
      when /^\d+$/ 
        reindex_options[:batch_size] = args[:batch_size].to_i if args[:batch_size].to_i > 0
      end

      puts "#{Parallel.processor_count} procesor(s)"
      puts "reindex using #{reindex_options[:exec_processor_size]} procesor(s)"

      Dir.glob(Rails.root.join('app/models/**/*.rb')).each { |path| require path }
      sunspot_models = Sunspot.searchable
      total_documents = sunspot_models.map { | m | m.count }.sum
      sorted_models = sunspot_models.sort{|a, b| a.count <=> b.count} 

      # Set up progress_bar to, ah, report progress
      begin
        require 'progress_bar'
        reindex_options[:progress_bar] = ProgressBar.new(total_documents)
      rescue LoadError => e
        $stdout.puts "Skipping progress bar: for progress reporting, add gem 'progress_bar' to your Gemfile"
      rescue Exception => e
        $stderr.puts "Error using progress bar: #{e.message}"
      end

      $stdout.puts "Reindex model list:"
      sorted_models.each { |model|
        $stdout.puts " name=#{model.name} record_size=#{model.count}"
      }
      $stdout.flush

      sorted_models.each do |model|
        model.solr_remove_all_from_index
        model.solr_index_parallel(reindex_options)
      end
    end

    desc "Reindex a manifestations"
    task :manifestations => :environment do
      request_reindex Manifestation 
    end

    desc "Reindex a items"
    task :items => :environment do
      request_reindex Item
    end
{'fraction_non_alphanumeric': 0.10640460860725177, 'fraction_numerical': 0.005083022704168078, 'mean_word_length': 2.4599452911293476, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2605472', 'n_tokens_mistral': 2783, 'n_tokens_neox': 2682, 'n_words': 580}