2018-06-27 03:31:41 -04:00
|
|
|
# frozen_string_literal: true
|
|
|
|
|
2018-02-26 06:17:50 -05:00
|
|
|
module WaitableWorker
|
|
|
|
extend ActiveSupport::Concern
|
|
|
|
|
2018-08-27 08:35:31 -04:00
|
|
|
class_methods do
|
2018-02-26 06:17:50 -05:00
|
|
|
# Schedules multiple jobs and waits for them to be completed.
|
2018-02-26 07:34:19 -05:00
|
|
|
def bulk_perform_and_wait(args_list, timeout: 10)
|
2018-02-26 06:17:50 -05:00
|
|
|
# Short-circuit: it's more efficient to do small numbers of jobs inline
|
|
|
|
return bulk_perform_inline(args_list) if args_list.size <= 3
|
|
|
|
|
2020-03-20 14:09:31 -04:00
|
|
|
# Don't wait if there's too many jobs to be waited for. Not including the
|
|
|
|
# waiter allows them to be deduplicated and it skips waiting for jobs that
|
|
|
|
# are not likely to finish within the timeout. This assumes we can process
|
|
|
|
# 10 jobs per second:
|
|
|
|
# https://gitlab.com/gitlab-com/gl-infra/scalability/-/issues/205
|
2020-03-30 23:07:51 -04:00
|
|
|
return bulk_perform_async(args_list) if args_list.length >= 10 * timeout
|
2020-03-20 14:09:31 -04:00
|
|
|
|
2020-02-25 16:09:23 -05:00
|
|
|
waiter = Gitlab::JobWaiter.new(args_list.size, worker_label: self.to_s)
|
2018-02-26 06:17:50 -05:00
|
|
|
|
|
|
|
# Point all the bulk jobs at the same JobWaiter. Converts, [[1], [2], [3]]
|
|
|
|
# into [[1, "key"], [2, "key"], [3, "key"]]
|
|
|
|
waiting_args_list = args_list.map { |args| [*args, waiter.key] }
|
|
|
|
bulk_perform_async(waiting_args_list)
|
|
|
|
|
2018-02-26 07:34:19 -05:00
|
|
|
waiter.wait(timeout)
|
2018-02-26 06:17:50 -05:00
|
|
|
end
|
|
|
|
|
|
|
|
# Performs multiple jobs directly. Failed jobs will be put into sidekiq so
|
|
|
|
# they can benefit from retries
|
|
|
|
def bulk_perform_inline(args_list)
|
|
|
|
failed = []
|
|
|
|
|
|
|
|
args_list.each do |args|
|
2019-03-13 09:42:43 -04:00
|
|
|
new.perform(*args)
|
|
|
|
rescue
|
|
|
|
failed << args
|
2018-02-26 06:17:50 -05:00
|
|
|
end
|
|
|
|
|
|
|
|
bulk_perform_async(failed) if failed.present?
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
|
|
|
def perform(*args)
|
|
|
|
notify_key = args.pop if Gitlab::JobWaiter.key?(args.last)
|
|
|
|
|
|
|
|
super(*args)
|
|
|
|
ensure
|
|
|
|
Gitlab::JobWaiter.notify(notify_key, jid) if notify_key
|
|
|
|
end
|
|
|
|
end
|