e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksaclass OntologyParsingWorker < BaseWorker
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa MAX_CONCURRENT_TRIES = 3
f1574e9c8462ac75b42c035513754fa4d0b6d1abTom Gehrke
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa sidekiq_options queue: 'hets'
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa # We cannot use Sidekiq's retry feature here because we need to retry per
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa # OntologyVersion in the queue of OntologyVersions
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa sidekiq_options retry: false
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa # queue is an array of triples like
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa # [ontology_version_id,
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa # {fast_parse: true/false, files_to_parse_afterwards: array of strings},
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa # try_count]
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa # where try_count is the number of the next evaluation attempt
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa def perform(version_with_options_queue)
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa return if version_with_options_queue.empty?
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa initialize_data(version_with_options_queue)
49f4112620975dae84bcd13f02667b00f16a3448Eugen Kuksa
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa timeout_job_id = ::TimeoutWorker.start_timeout_clock(@version_id)
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa parse_version
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa rescue Hets::ConcurrentEvaluator::AlreadyEvaluatingError
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa @concurrency_issue_handled = handle_concurrency_issue
ef6e04cd2f21652b42d621a4cfddb58010329063Tom Gehrke rescue HetsInstance::NoSelectableHetsInstanceError => e
ef6e04cd2f21652b42d621a4cfddb58010329063Tom Gehrke raise Sidekiq::Retries::Retry.new(e)
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa ensure
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa Sidekiq::Status.unschedule(timeout_job_id) if timeout_job_id
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa if @version_with_options_queue.tail.any? && !@concurrency_issue_handled
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa self.class.perform_async(@version_with_options_queue.tail)
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa end
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa end
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa protected
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa def initialize_data(version_with_options_queue)
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa @version_with_options_queue = version_with_options_queue
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa @version_id, @options, @try_count = @version_with_options_queue.head
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa @try_count ||= 1
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa @concurrency_issue_handled = false
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa end
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa def parse_version
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa version = OntologyVersion.find(@version_id)
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa @options.each do |method_name, value|
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa version.send(:"#{method_name}=", value)
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa end
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa version.parse
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa end
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa def handle_concurrency_issue
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa # At this point, it was already tried, hence we include equality
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa if @try_count >= MAX_CONCURRENT_TRIES
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa SequentialOntologyParsingWorker.perform_async(@version_id, @options)
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa else
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa self.class.perform_async(rotate_queue_and_increment_try_count)
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa end
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa true
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa end
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa def rotate_queue_and_increment_try_count
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa version_with_next_options = [@version_id, @options, @try_count + 1]
e8ec59dda8f903e81cb5cceaef93d20a0812fc77Eugen Kuksa [*@version_with_options_queue.tail, version_with_next_options]
49f4112620975dae84bcd13f02667b00f16a3448Eugen Kuksa end
49f4112620975dae84bcd13f02667b00f16a3448Eugen Kuksaend