cli_helper.rb 2.5 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485
  1. # frozen_string_literal: true
  2. dev_null = Logger.new('/dev/null')
  3. Rails.logger = dev_null
  4. ActiveRecord::Base.logger = dev_null
  5. ActiveJob::Base.logger = dev_null
  6. HttpLog.configuration.logger = dev_null
  7. Paperclip.options[:log] = false
  8. Chewy.logger = dev_null
  9. module Mastodon
  10. module CLIHelper
  11. def dry_run?
  12. options[:dry_run]
  13. end
  14. def create_progress_bar(total = nil)
  15. ProgressBar.create(total: total, format: '%c/%u |%b%i| %e')
  16. end
  17. def reset_connection_pools!
  18. ActiveRecord::Base.establish_connection(ActiveRecord::Base.configurations[Rails.env].dup.tap { |config| config['pool'] = options[:concurrency] + 1 })
  19. RedisConfiguration.establish_pool(options[:concurrency])
  20. end
  21. def parallelize_with_progress(scope)
  22. if options[:concurrency] < 1
  23. say('Cannot run with this concurrency setting, must be at least 1', :red)
  24. exit(1)
  25. end
  26. reset_connection_pools!
  27. progress = create_progress_bar(scope.count)
  28. pool = Concurrent::FixedThreadPool.new(options[:concurrency])
  29. total = Concurrent::AtomicFixnum.new(0)
  30. aggregate = Concurrent::AtomicFixnum.new(0)
  31. scope.reorder(nil).find_in_batches do |items|
  32. futures = []
  33. items.each do |item|
  34. futures << Concurrent::Future.execute(executor: pool) do
  35. begin
  36. if !progress.total.nil? && progress.progress + 1 > progress.total
  37. # The number of items has changed between start and now,
  38. # since there is no good way to predict the final count from
  39. # here, just change the progress bar to an indeterminate one
  40. progress.total = nil
  41. end
  42. progress.log("Processing #{item.id}") if options[:verbose]
  43. result = ActiveRecord::Base.connection_pool.with_connection do
  44. yield(item)
  45. ensure
  46. RedisConfiguration.pool.checkin if Thread.current[:redis]
  47. Thread.current[:redis] = nil
  48. end
  49. aggregate.increment(result) if result.is_a?(Integer)
  50. rescue => e
  51. progress.log pastel.red("Error processing #{item.id}: #{e}")
  52. ensure
  53. progress.increment
  54. end
  55. end
  56. end
  57. total.increment(items.size)
  58. futures.map(&:value)
  59. end
  60. progress.stop
  61. [total.value, aggregate.value]
  62. end
  63. def pastel
  64. @pastel ||= Pastel.new
  65. end
  66. end
  67. end