You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

83 lines
2.4 KiB

  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. if !progress.total.nil? && progress.progress + 1 > progress.total
  36. # The number of items has changed between start and now,
  37. # since there is no good way to predict the final count from
  38. # here, just change the progress bar to an indeterminate one
  39. progress.total = nil
  40. end
  41. progress.log("Processing #{item.id}") if options[:verbose]
  42. result = ActiveRecord::Base.connection_pool.with_connection do
  43. yield(item)
  44. ensure
  45. RedisConfiguration.pool.checkin if Thread.current[:redis]
  46. Thread.current[:redis] = nil
  47. end
  48. aggregate.increment(result) if result.is_a?(Integer)
  49. rescue => e
  50. progress.log pastel.red("Error processing #{item.id}: #{e}")
  51. ensure
  52. progress.increment
  53. end
  54. end
  55. total.increment(items.size)
  56. futures.map(&:value)
  57. end
  58. progress.stop
  59. [total.value, aggregate.value]
  60. end
  61. def pastel
  62. @pastel ||= Pastel.new
  63. end
  64. end
  65. end