From ce9df2fa8295a3bdd0da583ba5d0d90251e1d448 Mon Sep 17 00:00:00 2001 From: Eugen Rochko Date: Tue, 4 Apr 2017 13:01:14 +0200 Subject: Optimize filter methods in FeedManager a bit, use redis pipelining on merge/unmerge feed methods, do not re-create a dynamic class on each feed push call, make sure redis-rb uses hiredis --- app/lib/feed_manager.rb | 72 ++++++++++++++++++++++--------------------------- 1 file changed, 32 insertions(+), 40 deletions(-) (limited to 'app/lib/feed_manager.rb') diff --git a/app/lib/feed_manager.rb b/app/lib/feed_manager.rb index cd6ca1291..2c29275c8 100644 --- a/app/lib/feed_manager.rb +++ b/app/lib/feed_manager.rb @@ -51,9 +51,11 @@ class FeedManager def merge_into_timeline(from_account, into_account) timeline_key = key(:home, into_account.id) - from_account.statuses.limit(MAX_ITEMS).each do |status| - next if status.direct_visibility? || filter?(:home, status, into_account) - redis.zadd(timeline_key, status.id, status.id) + redis.pipelined do + from_account.statuses.limit(MAX_ITEMS).each do |status| + next if status.direct_visibility? || filter?(:home, status, into_account) + redis.zadd(timeline_key, status.id, status.id) + end end trim(:home, into_account.id) @@ -62,30 +64,18 @@ class FeedManager def unmerge_from_timeline(from_account, into_account) timeline_key = key(:home, into_account.id) - from_account.statuses.select('id').find_each do |status| - redis.zrem(timeline_key, status.id) - redis.zremrangebyscore(timeline_key, status.id, status.id) + from_account.statuses.select('id').find_in_batches do |statuses| + redis.pipelined do + statuses.each do |status| + redis.zrem(timeline_key, status.id) + redis.zremrangebyscore(timeline_key, status.id, status.id) + end + end end end def inline_render(target_account, template, object) - rabl_scope = Class.new do - include RoutingHelper - - def initialize(account) - @account = account - end - - def current_user - @account.try(:user) - end - - def current_account - @account - end - end - - Rabl::Renderer.new(template, object, view_path: 'app/views', format: :json, scope: rabl_scope.new(target_account)).render + Rabl::Renderer.new(template, object, view_path: 'app/views', format: :json, scope: InlineRablScope.new(target_account)).render end private @@ -95,37 +85,39 @@ class FeedManager end def filter_from_home?(status, receiver) - return true if receiver.muting?(status.account) + return true if status.reply? && status.in_reply_to_id.nil? + + check_for_mutes = [status.account_id] + check_for_mutes.concat([status.reblog.account_id]) if status.reblog? + + return true if receiver.muting?(check_for_mutes) - should_filter = false + check_for_blocks = status.mentions.map(&:account_id) + check_for_blocks.concat([status.reblog.account_id]) if status.reblog? - if status.reply? && status.in_reply_to_id.nil? - should_filter = true - elsif status.reply? && !status.in_reply_to_account_id.nil? # Filter out if it's a reply + return true if receiver.blocking?(check_for_blocks) + + if status.reply? && !status.in_reply_to_account_id.nil? # Filter out if it's a reply should_filter = !receiver.following?(status.in_reply_to_account) # and I'm not following the person it's a reply to should_filter &&= !(receiver.id == status.in_reply_to_account_id) # and it's not a reply to me should_filter &&= !(status.account_id == status.in_reply_to_account_id) # and it's not a self-reply + return should_filter elsif status.reblog? # Filter out a reblog - should_filter = receiver.blocking?(status.reblog.account) # if I'm blocking the reblogged person - should_filter ||= receiver.muting?(status.reblog.account) # or muting that person - should_filter ||= status.reblog.account.blocking?(receiver) # or if the author of the reblogged status is blocking me + return status.reblog.account.blocking?(receiver) # or if the author of the reblogged status is blocking me end - should_filter ||= receiver.blocking?(status.mentions.map(&:account_id)) # or if it mentions someone I blocked - - should_filter + false end def filter_from_mentions?(status, receiver) + check_for_blocks = [status.account_id] + check_for_blocks.concat(status.mentions.select('account_id').map(&:account_id)) + check_for_blocks.concat([status.in_reply_to_account]) if status.reply? && !status.in_reply_to_account_id.nil? + should_filter = receiver.id == status.account_id # Filter if I'm mentioning myself - should_filter ||= receiver.blocking?(status.account) # or it's from someone I blocked - should_filter ||= receiver.blocking?(status.mentions.includes(:account).map(&:account)) # or if it mentions someone I blocked + should_filter ||= receiver.blocking?(check_for_blocks) # or it's from someone I blocked, in reply to someone I blocked, or mentioning someone I blocked should_filter ||= (status.account.silenced? && !receiver.following?(status.account)) # of if the account is silenced and I'm not following them - if status.reply? && !status.in_reply_to_account_id.nil? # or it's a reply - should_filter ||= receiver.blocking?(status.in_reply_to_account) # to a user I blocked - end - should_filter end end -- cgit From b1f3499c3806682375a0496f99b4bc908d89cd84 Mon Sep 17 00:00:00 2001 From: Eugen Rochko Date: Tue, 4 Apr 2017 13:43:36 +0200 Subject: Optimize FeedManager#unmerge, and slightly optimize FeedManager#merge --- app/lib/feed_manager.rb | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) (limited to 'app/lib/feed_manager.rb') diff --git a/app/lib/feed_manager.rb b/app/lib/feed_manager.rb index 2c29275c8..919bc3df9 100644 --- a/app/lib/feed_manager.rb +++ b/app/lib/feed_manager.rb @@ -50,9 +50,15 @@ class FeedManager def merge_into_timeline(from_account, into_account) timeline_key = key(:home, into_account.id) + query = from_account.statuses.limit(MAX_ITEMS) + + if redis.zcard(timeline_key) >= FeedManager::MAX_ITEMS + oldest_home_score = redis.zrange(timeline_key, 0, 0, with_scores: true)&.first&.last&.to_i || 0 + query = query.where('id > ?', oldest_home_score) + end redis.pipelined do - from_account.statuses.limit(MAX_ITEMS).each do |status| + query.each do |status| next if status.direct_visibility? || filter?(:home, status, into_account) redis.zadd(timeline_key, status.id, status.id) end @@ -63,8 +69,9 @@ class FeedManager def unmerge_from_timeline(from_account, into_account) timeline_key = key(:home, into_account.id) + oldest_home_score = redis.zrange(timeline_key, 0, 0, with_scores: true)&.first&.last&.to_i || 0 - from_account.statuses.select('id').find_in_batches do |statuses| + from_account.statuses.select('id').where('id > ?', oldest_home_score).find_in_batches do |statuses| redis.pipelined do statuses.each do |status| redis.zrem(timeline_key, status.id) -- cgit From 82aaedec467815c2947a11651d5216bb88ce4038 Mon Sep 17 00:00:00 2001 From: Eugen Rochko Date: Tue, 4 Apr 2017 13:58:34 +0200 Subject: Reduce number of items in feeds, optimize regeneration worker slightly, make regeneration worker unique, (only schedule/execute once at a time) --- Gemfile | 2 ++ Gemfile.lock | 9 +++++++++ app/lib/feed_manager.rb | 6 +++--- app/services/precompute_feed_service.rb | 8 +++++--- app/workers/regeneration_worker.rb | 2 +- 5 files changed, 20 insertions(+), 7 deletions(-) (limited to 'app/lib/feed_manager.rb') diff --git a/Gemfile b/Gemfile index cb9824131..41c636904 100644 --- a/Gemfile +++ b/Gemfile @@ -46,6 +46,8 @@ gem 'will_paginate' gem 'rack-attack' gem 'rack-cors', require: 'rack/cors' gem 'sidekiq' +gem 'sidekiq-unique-jobs' +gem 'sidekiq-merger' gem 'rails-settings-cached' gem 'simple-navigation' gem 'statsd-instrument' diff --git a/Gemfile.lock b/Gemfile.lock index 6e3115249..27de1bee0 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -387,6 +387,13 @@ GEM connection_pool (~> 2.2, >= 2.2.0) rack-protection (>= 1.5.0) redis (~> 3.2, >= 3.2.1) + sidekiq-merger (0.0.11) + activesupport (>= 3.2, < 6) + concurrent-ruby (~> 1.0) + sidekiq (>= 3.4, < 5) + sidekiq-unique-jobs (4.0.18) + sidekiq (>= 2.6) + thor simple-navigation (4.0.3) activesupport (>= 2.3.2) simple_form (3.2.1) @@ -510,6 +517,8 @@ DEPENDENCIES sass-rails (~> 5.0) sdoc (~> 0.4.0) sidekiq + sidekiq-merger + sidekiq-unique-jobs simple-navigation simple_form simplecov diff --git a/app/lib/feed_manager.rb b/app/lib/feed_manager.rb index 919bc3df9..a2efcce10 100644 --- a/app/lib/feed_manager.rb +++ b/app/lib/feed_manager.rb @@ -5,7 +5,7 @@ require 'singleton' class FeedManager include Singleton - MAX_ITEMS = 800 + MAX_ITEMS = 400 def key(type, id) "feed:#{type}:#{id}" @@ -50,9 +50,9 @@ class FeedManager def merge_into_timeline(from_account, into_account) timeline_key = key(:home, into_account.id) - query = from_account.statuses.limit(MAX_ITEMS) + query = from_account.statuses.limit(FeedManager::MAX_ITEMS / 4) - if redis.zcard(timeline_key) >= FeedManager::MAX_ITEMS + if redis.zcard(timeline_key) >= FeedManager::MAX_ITEMS / 4 oldest_home_score = redis.zrange(timeline_key, 0, 0, with_scores: true)&.first&.last&.to_i || 0 query = query.where('id > ?', oldest_home_score) end diff --git a/app/services/precompute_feed_service.rb b/app/services/precompute_feed_service.rb index e1ec56e8d..a57c401d0 100644 --- a/app/services/precompute_feed_service.rb +++ b/app/services/precompute_feed_service.rb @@ -5,9 +5,11 @@ class PrecomputeFeedService < BaseService # @param [Symbol] type :home or :mentions # @param [Account] account def call(_, account) - Status.as_home_timeline(account).limit(FeedManager::MAX_ITEMS).each do |status| - next if status.direct_visibility? || FeedManager.instance.filter?(:home, status, account) - redis.zadd(FeedManager.instance.key(:home, account.id), status.id, status.reblog? ? status.reblog_of_id : status.id) + redis.pipelined do + Status.as_home_timeline(account).limit(FeedManager::MAX_ITEMS / 4).each do |status| + next if status.direct_visibility? || FeedManager.instance.filter?(:home, status, account) + redis.zadd(FeedManager.instance.key(:home, account.id), status.id, status.reblog? ? status.reblog_of_id : status.id) + end end end diff --git a/app/workers/regeneration_worker.rb b/app/workers/regeneration_worker.rb index 82665b581..da8b845f6 100644 --- a/app/workers/regeneration_worker.rb +++ b/app/workers/regeneration_worker.rb @@ -3,7 +3,7 @@ class RegenerationWorker include Sidekiq::Worker - sidekiq_options queue: 'pull', backtrace: true + sidekiq_options queue: 'pull', backtrace: true, unique: :until_executed def perform(account_id, _ = :home) PrecomputeFeedService.new.call(:home, Account.find(account_id)) -- cgit From 731e650681004bcb8ad11d610e017975a706f57d Mon Sep 17 00:00:00 2001 From: Kurtis Rainbolt-Greene Date: Tue, 4 Apr 2017 09:04:07 -0700 Subject: Use active record shorthand --- app/lib/feed_manager.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'app/lib/feed_manager.rb') diff --git a/app/lib/feed_manager.rb b/app/lib/feed_manager.rb index a2efcce10..9398d6c70 100644 --- a/app/lib/feed_manager.rb +++ b/app/lib/feed_manager.rb @@ -118,7 +118,7 @@ class FeedManager def filter_from_mentions?(status, receiver) check_for_blocks = [status.account_id] - check_for_blocks.concat(status.mentions.select('account_id').map(&:account_id)) + check_for_blocks.concat(status.mentions.pluck(:account_id)) check_for_blocks.concat([status.in_reply_to_account]) if status.reply? && !status.in_reply_to_account_id.nil? should_filter = receiver.id == status.account_id # Filter if I'm mentioning myself -- cgit From 6fd865c0004efbf11ee87c06fea9f48af567fabe Mon Sep 17 00:00:00 2001 From: Eugen Rochko Date: Tue, 4 Apr 2017 19:21:37 +0200 Subject: Spawn FeedInsertWorker to deliver status into personal feed --- app/lib/feed_manager.rb | 32 ++++++++++++++++---------------- app/services/fan_out_on_write_service.rb | 13 ++++++------- app/services/notify_service.rb | 2 +- app/services/precompute_feed_service.rb | 2 +- app/workers/feed_insert_worker.rb | 15 +++++++++++++++ 5 files changed, 39 insertions(+), 25 deletions(-) create mode 100644 app/workers/feed_insert_worker.rb (limited to 'app/lib/feed_manager.rb') diff --git a/app/lib/feed_manager.rb b/app/lib/feed_manager.rb index a2efcce10..28e712704 100644 --- a/app/lib/feed_manager.rb +++ b/app/lib/feed_manager.rb @@ -11,11 +11,11 @@ class FeedManager "feed:#{type}:#{id}" end - def filter?(timeline_type, status, receiver) + def filter?(timeline_type, status, receiver_id) if timeline_type == :home - filter_from_home?(status, receiver) + filter_from_home?(status, receiver_id) elsif timeline_type == :mentions - filter_from_mentions?(status, receiver) + filter_from_mentions?(status, receiver_id) else false end @@ -91,39 +91,39 @@ class FeedManager Redis.current end - def filter_from_home?(status, receiver) + def filter_from_home?(status, receiver_id) return true if status.reply? && status.in_reply_to_id.nil? check_for_mutes = [status.account_id] check_for_mutes.concat([status.reblog.account_id]) if status.reblog? - return true if receiver.muting?(check_for_mutes) + return true if Mute.where(account_id: receiver_id, target_account_id: check_for_mutes).any? check_for_blocks = status.mentions.map(&:account_id) check_for_blocks.concat([status.reblog.account_id]) if status.reblog? - return true if receiver.blocking?(check_for_blocks) + return true if Block.where(account_id: receiver_id, target_account_id: check_for_blocks).any? - if status.reply? && !status.in_reply_to_account_id.nil? # Filter out if it's a reply - should_filter = !receiver.following?(status.in_reply_to_account) # and I'm not following the person it's a reply to - should_filter &&= !(receiver.id == status.in_reply_to_account_id) # and it's not a reply to me - should_filter &&= !(status.account_id == status.in_reply_to_account_id) # and it's not a self-reply + if status.reply? && !status.in_reply_to_account_id.nil? # Filter out if it's a reply + should_filter = !Follow.where(account_id: receiver_id, target_account_id: status.in_reply_to_account_id).exists? # and I'm not following the person it's a reply to + should_filter &&= !(receiver_id == status.in_reply_to_account_id) # and it's not a reply to me + should_filter &&= !(status.account_id == status.in_reply_to_account_id) # and it's not a self-reply return should_filter - elsif status.reblog? # Filter out a reblog - return status.reblog.account.blocking?(receiver) # or if the author of the reblogged status is blocking me + elsif status.reblog? # Filter out a reblog + return Block.where(account_id: status.reblog.account_id, target_account_id: receiver_id).exists? # or if the author of the reblogged status is blocking me end false end - def filter_from_mentions?(status, receiver) + def filter_from_mentions?(status, receiver_id) check_for_blocks = [status.account_id] check_for_blocks.concat(status.mentions.select('account_id').map(&:account_id)) check_for_blocks.concat([status.in_reply_to_account]) if status.reply? && !status.in_reply_to_account_id.nil? - should_filter = receiver.id == status.account_id # Filter if I'm mentioning myself - should_filter ||= receiver.blocking?(check_for_blocks) # or it's from someone I blocked, in reply to someone I blocked, or mentioning someone I blocked - should_filter ||= (status.account.silenced? && !receiver.following?(status.account)) # of if the account is silenced and I'm not following them + should_filter = receiver_id == status.account_id # Filter if I'm mentioning myself + should_filter ||= Block.where(account_id: receiver_id, target_account_id: check_for_blocks).any? # or it's from someone I blocked, in reply to someone I blocked, or mentioning someone I blocked + should_filter ||= (status.account.silenced? && !Follow.where(account_id: receiver_id, target_account_id: status.account_id).exists?) # of if the account is silenced and I'm not following them should_filter end diff --git a/app/services/fan_out_on_write_service.rb b/app/services/fan_out_on_write_service.rb index df404cbef..42222c25b 100644 --- a/app/services/fan_out_on_write_service.rb +++ b/app/services/fan_out_on_write_service.rb @@ -33,9 +33,8 @@ class FanOutOnWriteService < BaseService def deliver_to_followers(status) Rails.logger.debug "Delivering status #{status.id} to followers" - status.account.followers.where(domain: nil).joins(:user).where('users.current_sign_in_at > ?', 14.days.ago).find_each do |follower| - next if FeedManager.instance.filter?(:home, status, follower) - FeedManager.instance.push(:home, follower, status) + status.account.followers.where(domain: nil).joins(:user).where('users.current_sign_in_at > ?', 14.days.ago).select(:id).find_each do |follower| + FeedInsertWorker.perform_async(status.id, follower.id) end end @@ -44,7 +43,7 @@ class FanOutOnWriteService < BaseService status.mentions.includes(:account).each do |mention| mentioned_account = mention.account - next if !mentioned_account.local? || !mentioned_account.following?(status.account) || FeedManager.instance.filter?(:home, status, mentioned_account) + next if !mentioned_account.local? || !mentioned_account.following?(status.account) || FeedManager.instance.filter?(:home, status, mention.account_id) FeedManager.instance.push(:home, mentioned_account, status) end end @@ -54,9 +53,9 @@ class FanOutOnWriteService < BaseService payload = FeedManager.instance.inline_render(nil, 'api/v1/statuses/show', status) - status.tags.find_each do |tag| - FeedManager.instance.broadcast("hashtag:#{tag.name}", event: 'update', payload: payload) - FeedManager.instance.broadcast("hashtag:#{tag.name}:local", event: 'update', payload: payload) if status.account.local? + status.tags.pluck(:name).each do |hashtag| + FeedManager.instance.broadcast("hashtag:#{hashtag}", event: 'update', payload: payload) + FeedManager.instance.broadcast("hashtag:#{hashtag}:local", event: 'update', payload: payload) if status.account.local? end end diff --git a/app/services/notify_service.rb b/app/services/notify_service.rb index 942cd9d21..24486f220 100644 --- a/app/services/notify_service.rb +++ b/app/services/notify_service.rb @@ -17,7 +17,7 @@ class NotifyService < BaseService private def blocked_mention? - FeedManager.instance.filter?(:mentions, @notification.mention.status, @recipient) + FeedManager.instance.filter?(:mentions, @notification.mention.status, @recipient.id) end def blocked_favourite? diff --git a/app/services/precompute_feed_service.rb b/app/services/precompute_feed_service.rb index a57c401d0..07dcb81da 100644 --- a/app/services/precompute_feed_service.rb +++ b/app/services/precompute_feed_service.rb @@ -7,7 +7,7 @@ class PrecomputeFeedService < BaseService def call(_, account) redis.pipelined do Status.as_home_timeline(account).limit(FeedManager::MAX_ITEMS / 4).each do |status| - next if status.direct_visibility? || FeedManager.instance.filter?(:home, status, account) + next if status.direct_visibility? || FeedManager.instance.filter?(:home, status, account.id) redis.zadd(FeedManager.instance.key(:home, account.id), status.id, status.reblog? ? status.reblog_of_id : status.id) end end diff --git a/app/workers/feed_insert_worker.rb b/app/workers/feed_insert_worker.rb new file mode 100644 index 000000000..a58dfaa74 --- /dev/null +++ b/app/workers/feed_insert_worker.rb @@ -0,0 +1,15 @@ +# frozen_string_literal: true + +class FeedInsertWorker + include Sidekiq::Worker + + def perform(status_id, follower_id) + status = Status.find(status_id) + follower = Account.find(follower_id) + + return if FeedManager.instance.filter?(:home, status, follower.id) + FeedManager.instance.push(:home, follower, status) + rescue ActiveRecord::RecordNotFound + true + end +end -- cgit