Merge pull request #389 from blackcandy-org/parallel-sync

Parallel media sync

Gemfile

@@ -67,6 +67,9 @@ gem "bcrypt", "~> 3.1.11"
 # For sync on library changes
 gem "listen", "~> 3.8.0"
 
+# For parallel media sync
+gem "parallel", "~> 1.25.0"
+
 # For daemonize library sync process
 gem "daemons", "~> 1.4.0"
 

Gemfile.lock

@@ -238,7 +238,7 @@ GEM
     oj (3.16.3)
       bigdecimal (>= 3.0)
     pagy (6.0.4)
-    parallel (1.24.0)
+    parallel (1.25.1)
     parser (3.3.1.0)
       ast (~> 2.4.1)
       racc
@@ -417,6 +417,7 @@ DEPENDENCIES
   litestack (~> 0.4.2)
   memory_profiler (~> 0.9.13)
   pagy (~> 6.0.0)
+  parallel (~> 1.25.0)
   pg (~> 1.5.4)
   propshaft (~> 0.7.0)
   puma (~> 6.4.0)

app/assets/stylesheets/components/_form.scss

@@ -22,6 +22,10 @@
   margin-bottom: spacing("narrow");
 }
 
+.c-form__field label[disabled] {
+  opacity: 0.5;
+}
+
 .c-form__field--inline label {
   margin-bottom: 0;
   margin-right: spacing("tiny");

app/controllers/application_controller.rb

@@ -109,7 +109,7 @@ def require_login
   end
 
   def require_admin
-    raise BlackCandy::Forbidden if BlackCandy::Config.demo_mode? || !is_admin?
+    raise BlackCandy::Forbidden if BlackCandy.config.demo_mode? || !is_admin?
   end
 
   def ios_app?
@@ -125,7 +125,7 @@ def render_json_error(error, status)
   end
 
   def send_local_file(file_path, format, nginx_headers: {})
-    if BlackCandy::Config.nginx_sendfile?
+    if BlackCandy.config.nginx_sendfile?
       nginx_headers.each { |name, value| response.headers[name] = value }
       send_file file_path
 

app/controllers/media_syncing_controller.rb

@@ -8,7 +8,7 @@ def create
       flash[:error] = t("error.syncing_in_progress")
       redirect_to setting_path
     else
-      MediaSyncJob.perform_later
+      MediaSyncAllJob.perform_later
     end
   end
 end

app/controllers/users_controller.rb

@@ -50,7 +50,7 @@ def find_user
   end
 
   def auth_user
-    raise BlackCandy::Forbidden if BlackCandy::Config.demo_mode?
+    raise BlackCandy::Forbidden if BlackCandy.config.demo_mode?
     raise BlackCandy::Forbidden unless @user == Current.user || is_admin?
   end
 

app/helpers/sessions_helper.rb

@@ -10,7 +10,7 @@ def is_admin?
   end
 
   def login(session)
-    cookies.signed[:session_id] = {value: session.id, expires: 1.year.from_now, httponly: true, secure: BlackCandy::Config.force_ssl?}
+    cookies.signed[:session_id] = {value: session.id, expires: 1.year.from_now, httponly: true, secure: BlackCandy.config.force_ssl?}
   end
 
   def logout

app/jobs/media_sync_all_job.rb

@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+class MediaSyncAllJob < MediaSyncJob
+  def perform(dir = Setting.media_path)
+    file_paths = MediaFile.file_paths(dir)
+    file_md5_hashes = Parallel.map(file_paths, in_processes: self.class.parallel_processor_count) do |file_path|
+      MediaFile.get_md5_hash(file_path, with_mtime: true)
+    end
+
+    existing_songs = Song.where(md5_hash: file_md5_hashes)
+    added_file_paths = file_paths - existing_songs.pluck(:file_path)
+    added_song_hashes = added_file_paths.blank? ? [] : parallel_sync(:added, added_file_paths).flatten.compact
+
+    Media.clean_up(added_song_hashes + existing_songs.pluck(:md5_hash))
+  end
+
+  private
+
+  def after_sync(_)
+    Media.instance.broadcast_render_to "media_sync", partial: "media_syncing/syncing", locals: {syncing: false}
+    super(fetch_external_metadata: true)
+  end
+end

app/jobs/media_sync_job.rb

@@ -1,14 +1,48 @@
+# frozen_string_literal: true
+
 class MediaSyncJob < ApplicationJob
+  include BlackCandy::Configurable
+
+  has_config :parallel_processor_count, default: Parallel.processor_count, env_prefix: "media_sync"
+
   queue_as :critical
 
   # Limits the concurrency to 1 to prevent inconsistent media syncing data.
   limits_concurrency to: 1, key: :media_sync
 
-  def perform(type = :all, file_paths = [])
-    if type == :all
-      Media.sync_all
-    else
-      Media.sync(type, file_paths)
+  before_perform :before_sync
+
+  after_perform do |job|
+    sync_type = job.arguments.first
+    after_sync(fetch_external_metadata: sync_type != :removed)
+  end
+
+  def perform(type, file_paths = [])
+    parallel_sync(type, file_paths)
+  end
+
+  def self.parallel_processor_count
+    return 0 unless Setting.enable_parallel_media_sync?
+    config.parallel_processor_count
+  end
+
+  private
+
+  def parallel_sync(type, file_paths)
+    parallel_processor_count = self.class.parallel_processor_count
+    grouped_file_paths = (parallel_processor_count > 0) ? file_paths.in_groups(parallel_processor_count, false).compact_blank : [file_paths]
+
+    Parallel.each grouped_file_paths, in_processes: parallel_processor_count do |paths|
+      Media.sync(type, paths)
     end
   end
+
+  def before_sync
+    Media.syncing = true
+  end
+
+  def after_sync(fetch_external_metadata: true)
+    Media.syncing = false
+    Media.fetch_external_metadata if fetch_external_metadata
+  end
 end

app/models/album.rb

@@ -11,7 +11,6 @@ class Album < ApplicationRecord
   after_initialize :set_default_name, if: :new_record?
 
   validates :name, presence: true
-  validates :name, uniqueness: {scope: :artist}
 
   has_many :songs, -> { order(:discnum, :tracknum) }, inverse_of: :album, dependent: :destroy
   belongs_to :artist, touch: true

app/models/media.rb

@@ -6,21 +6,10 @@ class Media
   extend ActiveModel::Naming
 
   class << self
-    def sync_all(dir = Setting.media_path)
-      sync(:all, MediaFile.file_paths(dir))
-    ensure
-      instance.broadcast_render_to "media_sync", partial: "media_syncing/syncing", locals: {syncing: false}
-    end
-
     def sync(type, file_paths = [])
-      self.syncing = true
-
       return if file_paths.blank?
 
       case type
-      when :all
-        file_hashes = add_files(file_paths)
-        clean_up(file_hashes)
       when :added
         add_files(file_paths)
       when :removed
@@ -29,10 +18,6 @@ def sync(type, file_paths = [])
         remove_files(file_paths)
         add_files(file_paths)
       end
-
-      fetch_external_metadata unless type == :removed
-    ensure
-      self.syncing = false
     end
 
     def syncing?
@@ -44,20 +29,39 @@ def syncing=(is_syncing)
       Rails.cache.write("media_syncing", is_syncing, expires_in: 1.hour)
     end
 
+    def clean_up(file_hashes = [])
+      Song.where.not(md5_hash: file_hashes).destroy_all if file_hashes.present?
+
+      # Clean up no content albums and artist.
+      Album.where.missing(:songs).destroy_all
+      Artist.where.missing(:songs, :albums).destroy_all
+    end
+
+    def fetch_external_metadata
+      return unless Setting.discogs_token.present?
+
+      jobs = []
+
+      Artist.lack_metadata.find_each do |artist|
+        jobs << AttachCoverImageFromDiscogsJob.new(artist)
+      end
+
+      Album.lack_metadata.find_each do |album|
+        jobs << AttachCoverImageFromDiscogsJob.new(album)
+      end
+
+      ActiveJob.perform_all_later(jobs)
+    end
+
     private
 
     def add_files(file_paths)
-      file_md5_hashes = file_paths.map { |file_path| MediaFile.get_md5_hash(file_path, with_mtime: true) }
-      existing_songs = Song.where(md5_hash: file_md5_hashes)
-
-      added_song_hashes = (file_paths - existing_songs.pluck(:file_path)).map do |file_path|
+      file_paths.map do |file_path|
         file_info = MediaFile.file_info(file_path)
         file_info[:md5_hash] if attach(file_info)
       rescue
         next
       end.compact
-
-      added_song_hashes + existing_songs.pluck(:md5_hash)
     end
 
     def remove_files(file_paths)
@@ -68,10 +72,10 @@ def remove_files(file_paths)
     end
 
     def attach(file_info)
-      artist = Artist.find_or_create_by!(name: file_info[:artist_name] || Artist::UNKNOWN_NAME)
-      various_artist = Artist.find_or_create_by!(various: true) if various_artist?(file_info)
+      artist = Artist.create_or_find_by!(name: file_info[:artist_name] || Artist::UNKNOWN_NAME)
+      various_artist = Artist.create_or_find_by!(various: true) if various_artist?(file_info)
 
-      album = Album.find_or_initialize_by(
+      album = Album.create_or_find_by!(
         artist_id: various_artist&.id || artist.id,
         name: file_info[:album_name] || Album::UNKNOWN_NAME
       )
@@ -82,8 +86,9 @@ def attach(file_info)
         album.cover_image.attach(file_info[:image]) if file_info[:image].present?
       end
 
-      song = Song.find_or_initialize_by(md5_hash: file_info[:md5_hash])
-      song.update!(song_info(file_info).merge(album_id: album.id, artist_id: artist.id))
+      Song.create_or_find_by!(md5_hash: file_info[:md5_hash]) do |item|
+        item.attributes = song_info(file_info).merge(album_id: album.id, artist_id: artist.id)
+      end
     end
 
     def song_info(file_info)
@@ -98,29 +103,5 @@ def various_artist?(file_info)
       albumartist = file_info[:albumartist_name]
       albumartist.present? && (albumartist.casecmp("various artists").zero? || albumartist != file_info[:artist_name])
     end
-
-    def clean_up(file_hashes = [])
-      Song.where.not(md5_hash: file_hashes).destroy_all if file_hashes.present?
-
-      # Clean up no content albums and artist.
-      Album.where.missing(:songs).destroy_all
-      Artist.where.missing(:songs, :albums).destroy_all
-    end
-
-    def fetch_external_metadata
-      return unless Setting.discogs_token.present?
-
-      jobs = []
-
-      Artist.lack_metadata.find_each do |artist|
-        jobs << AttachCoverImageFromDiscogsJob.new(artist)
-      end
-
-      Album.lack_metadata.find_each do |album|
-        jobs << AttachCoverImageFromDiscogsJob.new(album)
-      end
-
-      ActiveJob.perform_all_later(jobs)
-    end
   end
 end

app/models/media_listener.rb

@@ -2,35 +2,15 @@
 
 class MediaListener
   include Singleton
+  include BlackCandy::Configurable
 
   SERVICE_PATH = File.join(Rails.root, "lib", "daemons", "media_listener_service")
-  class Config
-    DEFAULTS = {
-      service_name: "media_listener_service",
-      pid_dir: File.join(Rails.root, "tmp", "pids")
-    }
-
-    attr_accessor :service_name, :pid_dir
-
-    def initialize
-      DEFAULTS.each do |key, value|
-        send("#{key}=", value)
-      end
-    end
-  end
+
+  has_config :service_name, default: "media_listener_service", env_prefix: "media_listener"
+  has_config :pid_dir, default: File.join(Rails.root, "tmp", "pids"), env_prefix: "media_listener"
 
   class << self
     delegate :start, :stop, :running?, to: :instance
-
-    def config
-      yield instance.config
-    end
-  end
-
-  attr_reader :config
-
-  def initialize
-    @config = Config.new
   end
 
   def start
@@ -42,7 +22,7 @@ def stop
   end
 
   def running?
-    pid_file_path = Daemons::PidFile.find_files(@config.pid_dir, @config.service_name).first
+    pid_file_path = Daemons::PidFile.find_files(self.class.config.pid_dir, self.class.config.service_name).first
     return false unless pid_file_path.present?
 
     pid_file = Daemons::PidFile.existing(pid_file_path)
@@ -52,6 +32,6 @@ def running?
   private
 
   def run(command)
-    system "bundle exec #{SERVICE_PATH} #{command} -d #{@config.pid_dir} -n #{@config.service_name}"
+    system "bundle exec #{SERVICE_PATH} #{command} -d #{self.class.config.pid_dir} -n #{self.class.config.service_name}"
   end
 end

app/models/setting.rb

@@ -5,14 +5,16 @@ class Setting < ApplicationRecord
 
   AVAILABLE_BITRATE_OPTIONS = [128, 192, 320].freeze
 
-  has_setting :media_path, default: proc { BlackCandy::Config.media_path }
+  has_setting :media_path, default: proc { BlackCandy.config.media_path }
   has_setting :discogs_token
   has_setting :transcode_bitrate, type: :integer, default: 128
   has_setting :allow_transcode_lossless, type: :boolean, default: false
   has_setting :enable_media_listener, type: :boolean, default: false
+  has_setting :enable_parallel_media_sync, type: :boolean, default: false
 
   validates :transcode_bitrate, inclusion: {in: AVAILABLE_BITRATE_OPTIONS}, allow_nil: true
   validate :media_path_exist
+  validate :parallel_media_sync_database
 
   after_update :toggle_media_listener, if: :saved_change_to_enable_media_listener?
   after_update_commit :sync_media, if: :saved_change_to_media_path?
@@ -30,6 +32,14 @@ def media_path_exist
     errors.add(:media_path, :unreadable) unless File.readable?(path)
   end
 
+  def parallel_media_sync_database
+    return unless enable_parallel_media_sync?
+
+    if BlackCandy.config.db_adapter == "sqlite"
+      errors.add(:enable_parallel_media_sync, :not_supported_with_sqlite)
+    end
+  end
+
   def sync_media
     MediaSyncJob.perform_later
   end