<commit_before>require 'backburner/job'

module Backburner
  #
  # @abstract Subclass and override {#process_tube_names}, {#prepare} and {#start} to implement
  #   a custom Worker class.
  #
  class Worker
    include Backburner::Helpers
    include Backburner::Logger

    # Backburner::Worker.known_queue_classes
    # List of known_queue_classes
    class << self
      attr_writer :known_queue_classes
      def known_queue_classes; @known_queue_classes ||= []; end
    end

    # Enqueues a job to be processed later by a worker.
    # Options: `pri` (priority), `delay` (delay in secs), `ttr` (time to respond), `queue` (queue name)
    #
    # @raise [Beaneater::NotConnected] If beanstalk fails to connect.
    # @example
    #   Backburner::Worker.enqueue NewsletterSender, [self.id, user.id], :ttr => 1000
    #
    def self.enqueue(job_class, args=[], opts={})
      pri   = opts[:pri] && (Proc === opts[:pri] ? opts[:pri].call(job_class, args) : opts[:pri])
      pri   = resolve_priority(pri || job_class)

      delay = [0, opts[:delay].to_i].max

      ttr   = opts[:ttr] && (Proc === opts[:ttr] ? opts[:ttr].call(job_class, args) : opts[:ttr])
      ttr   = resolve_respond_timeout(ttr || job_class)

      res   = Backburner::Hooks.invoke_hook_events(job_class, :before_enqueue, *args)

      return nil unless res # stop if hook is false

      data = { :class => job_class.name, :args => args }
      queue = opts[:queue] && (Proc === opts[:queue] ? opts[:queue].call(job_class) : opts[:queue])

      begin
        response = nil
        connection = Backburner::Connection.new(Backburner.configuration.beanstalk_url)
        connection.retryable do
          tube = connection.tubes[expand_tube_name(queue || job_class)]
          serialized_data = Backburner.configuration.job_serializer_proc.call(data)
          response = tube.put(serialized_data, :pri => pri, :delay => delay, :ttr => ttr)
        end
        return nil unless Backburner::Hooks.invoke_hook_events(job_class, :after_enqueue, *args)
      ensure
        connection.close if connection
      end

      response
    end

    # Starts processing jobs with the specified tube_names.
    #
    # @example
    #   Backburner::Worker.start(["foo.tube.name"])
    #
    def self.start(tube_names=nil)
      begin
        self.new(tube_names).start
      rescue SystemExit
        # do nothing
      end
    end

    # List of tube names to be watched and processed
    attr_accessor :tube_names, :connection

    # Constructs a new worker for processing jobs within specified tubes.
    #
    # @example
    #   Worker.new(['test.job'])
    def initialize(tube_names=nil)
      @connection = new_connection
      @tube_names = self.process_tube_names(tube_names)
      register_signal_handlers!
    end

    # Starts processing ready jobs indefinitely.
    # Primary way to consume and process jobs in specified tubes.
    #
    # @example
    #   @worker.start
    #
    def start
      raise NotImplementedError
    end

    # Used to prepare the job queues before job processing is initiated.
    #
    # @raise [Beaneater::NotConnected] If beanstalk fails to connect.
    # @example
    #   @worker.prepare
    #
    # @abstract Define this in your worker subclass
    # to be run once before processing. Recommended to watch tubes
    # or print a message to the logs with 'log_info'
    #
    def prepare
      raise NotImplementedError
    end

    # Triggers this worker to shutdown
    def shutdown
      Thread.new do
        log_info 'Worker exiting...'
      end
      Kernel.exit
    end

    # Processes tube_names given tube_names array.
    # Should return normalized tube_names as an array of strings.
    #
    # @example
    #   process_tube_names([['foo'], ['bar']])
    #   => ['foo', 'bar', 'baz']
    #
    # @note This method can be overridden in inherited workers
    # to add more complex tube name processing.
    def process_tube_names(tube_names)
      compact_tube_names(tube_names)
    end

    # Performs a job by reserving a job from beanstalk and processing it
    #
    # @example
    #   @worker.work_one_job
    # @raise [Beaneater::NotConnected] If beanstalk fails to connect multiple times.
    def work_one_job(conn = connection)
      begin
        job = reserve_job(conn)
      rescue Beaneater::TimedOutError => e
        return
      end

      self.log_job_begin(job.name, job.args)
      job.process
      self.log_job_end(job.name)

    rescue Backburner::Job::JobFormatInvalid => e
      self.log_error self.exception_message(e)
    rescue => e # Error occurred processing job
      self.log_error self.exception_message(e)

      unless job
        self.log_error "Error occurred before we were able to assign a job. Giving up without retrying!"
        return
      end

      # NB: There's a slight chance here that the connection to beanstalkd has
      # gone down between the time we reserved / processed the job and here.
      num_retries = job.stats.releases
      retry_status = "failed: attempt #{num_retries+1} of #{queue_config.max_job_retries+1}"
      if num_retries < queue_config.max_job_retries # retry again
        delay = queue_config.retry_delay_proc.call(queue_config.retry_delay, num_retries) rescue queue_config.retry_delay
        job.retry(num_retries + 1, delay)
        self.log_job_end(job.name, "#{retry_status}, retrying in #{delay}s") if job_started_at
      else # retries failed, bury
        job.bury
        self.log_job_end(job.name, "#{retry_status}, burying") if job_started_at
      end<commit_msg>Added FIXME for job.retry leading to NOT_FOUND exception<commit_after>require 'backburner/job'

module Backburner
  #
  # @abstract Subclass and override {#process_tube_names}, {#prepare} and {#start} to implement
  #   a custom Worker class.
  #
  class Worker
    include Backburner::Helpers
    include Backburner::Logger

    # Backburner::Worker.known_queue_classes
    # List of known_queue_classes
    class << self
      attr_writer :known_queue_classes
      def known_queue_classes; @known_queue_classes ||= []; end
    end

    # Enqueues a job to be processed later by a worker.
    # Options: `pri` (priority), `delay` (delay in secs), `ttr` (time to respond), `queue` (queue name)
    #
    # @raise [Beaneater::NotConnected] If beanstalk fails to connect.
    # @example
    #   Backburner::Worker.enqueue NewsletterSender, [self.id, user.id], :ttr => 1000
    #
    def self.enqueue(job_class, args=[], opts={})
      pri   = opts[:pri] && (Proc === opts[:pri] ? opts[:pri].call(job_class, args) : opts[:pri])
      pri   = resolve_priority(pri || job_class)

      delay = [0, opts[:delay].to_i].max

      ttr   = opts[:ttr] && (Proc === opts[:ttr] ? opts[:ttr].call(job_class, args) : opts[:ttr])
      ttr   = resolve_respond_timeout(ttr || job_class)

      res   = Backburner::Hooks.invoke_hook_events(job_class, :before_enqueue, *args)

      return nil unless res # stop if hook is false

      data = { :class => job_class.name, :args => args }
      queue = opts[:queue] && (Proc === opts[:queue] ? opts[:queue].call(job_class) : opts[:queue])

      begin
        response = nil
        connection = Backburner::Connection.new(Backburner.configuration.beanstalk_url)
        connection.retryable do
          tube = connection.tubes[expand_tube_name(queue || job_class)]
          serialized_data = Backburner.configuration.job_serializer_proc.call(data)
          response = tube.put(serialized_data, :pri => pri, :delay => delay, :ttr => ttr)
        end
        return nil unless Backburner::Hooks.invoke_hook_events(job_class, :after_enqueue, *args)
      ensure
        connection.close if connection
      end

      response
    end

    # Starts processing jobs with the specified tube_names.
    #
    # @example
    #   Backburner::Worker.start(["foo.tube.name"])
    #
    def self.start(tube_names=nil)
      begin
        self.new(tube_names).start
      rescue SystemExit
        # do nothing
      end
    end

    # List of tube names to be watched and processed
    attr_accessor :tube_names, :connection

    # Constructs a new worker for processing jobs within specified tubes.
    #
    # @example
    #   Worker.new(['test.job'])
    def initialize(tube_names=nil)
      @connection = new_connection
      @tube_names = self.process_tube_names(tube_names)
      register_signal_handlers!
    end

    # Starts processing ready jobs indefinitely.
    # Primary way to consume and process jobs in specified tubes.
    #
    # @example
    #   @worker.start
    #
    def start
      raise NotImplementedError
    end

    # Used to prepare the job queues before job processing is initiated.
    #
    # @raise [Beaneater::NotConnected] If beanstalk fails to connect.
    # @example
    #   @worker.prepare
    #
    # @abstract Define this in your worker subclass
    # to be run once before processing. Recommended to watch tubes
    # or print a message to the logs with 'log_info'
    #
    def prepare
      raise NotImplementedError
    end

    # Triggers this worker to shutdown
    def shutdown
      Thread.new do
        log_info 'Worker exiting...'
      end
      Kernel.exit
    end

    # Processes tube_names given tube_names array.
    # Should return normalized tube_names as an array of strings.
    #
    # @example
    #   process_tube_names([['foo'], ['bar']])
    #   => ['foo', 'bar', 'baz']
    #
    # @note This method can be overridden in inherited workers
    # to add more complex tube name processing.
    def process_tube_names(tube_names)
      compact_tube_names(tube_names)
    end

    # Performs a job by reserving a job from beanstalk and processing it
    #
    # @example
    #   @worker.work_one_job
    # @raise [Beaneater::NotConnected] If beanstalk fails to connect multiple times.
    def work_one_job(conn = connection)
      begin
        job = reserve_job(conn)
      rescue Beaneater::TimedOutError => e
        return
      end

      self.log_job_begin(job.name, job.args)
      job.process
      self.log_job_end(job.name)

    rescue Backburner::Job::JobFormatInvalid => e
      self.log_error self.exception_message(e)
    rescue => e # Error occurred processing job
      self.log_error self.exception_message(e)

      unless job
        self.log_error "Error occurred before we were able to assign a job. Giving up without retrying!"
        return
      end

      # NB: There's a slight chance here that the connection to beanstalkd has
      # gone down between the time we reserved / processed the job and here.
      num_retries = job.stats.releases
      retry_status = "failed: attempt #{num_retries+1} of #{queue_config.max_job_retries+1}"
      if num_retries < queue_config.max_job_retries # retry again
        delay = queue_config.retry_delay_proc.call(queue_config.retry_delay, num_retries) rescue queue_config.retry_delay
        job.retry(num_retries + 1, delay)  # FIXME: If we're very close to the job-release time then this can fail (with a NOT_FOUND for the reservation)
        self.log_job_end(job.name, "#{retry_status}, retrying in #{delay}s") if job_started_at
      else # retries failed, bury
        job.bury
        self.log_job_end(job.name, "#{retry_status}, burying") if job_started_at
      end
{'fraction_non_alphanumeric': 0.0951243252809486, 'fraction_numerical': 0.0014158039111583046, 'mean_word_length': 2.8455256890098672, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '84772', 'n_tokens_mistral': 3527, 'n_tokens_neox': 3397, 'n_words': 969}