Class: GoodJob::Adapter

Inherits:
Object
  • Object
show all
Defined in:
lib/good_job/adapter.rb,
lib/good_job/adapter/inline_buffer.rb

Overview

ActiveJob Adapter.

Defined Under Namespace

Classes: InlineBuffer

Class Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(execution_mode: nil, _capsule: GoodJob.capsule) ⇒ Adapter

Returns a new instance of Adapter.

Parameters:

  • execution_mode (Symbol, nil) (defaults to: nil) —

    specifies how and where jobs should be executed. You can also set this with the environment variable GOOD_JOB_EXECUTION_MODE.

    • :inline executes jobs immediately in whatever process queued them (usually the web server process). This should only be used in test and development environments.

    • :external causes the adapter to enqueue jobs, but not execute them. When using this option (the default for production environments), you’ll need to use the command-line tool to actually execute your jobs.

    • :async (or :async_server) executes jobs in separate threads within the Rails web server process (‘bundle exec rails server`). It can be more economical for small workloads because you don’t need a separate machine or environment for running your jobs, but if your web server is under heavy load or your jobs require a lot of resources, you should choose :external instead. When not in the Rails web server, jobs will execute in :external mode to ensure jobs are not executed within ‘rails console`, `rails db:migrate`, `rails assets:prepare`, etc.

    • :async_all executes jobs in any Rails process.

    The default value depends on the Rails environment:

    • development: :async:

    -+test+: +:inline+

    • production and all other environments: :external



28
29
30
31
32
33
34
35
 
# File 'lib/good_job/adapter.rb', line 28

def initialize(execution_mode: nil, _capsule: GoodJob.capsule) # rubocop:disable Lint/UnderscorePrefixedVariableName
  @_execution_mode_override = execution_mode
  GoodJob::Configuration.validate_execution_mode(@_execution_mode_override) if @_execution_mode_override
  @capsule = _capsule

  start_async if GoodJob.async_ready?
  self.class.instances << self
end

Class Attribute Details

.instances ⇒ Array<GoodJob::Adapter>? (readonly)

List of all instantiated Adapters in the current process.

Returns:



12
 
# File 'lib/good_job/adapter.rb', line 12

cattr_reader :instances, default: Concurrent::Array.new, instance_reader: false

Instance Method Details

#async_started? ⇒ Boolean

Whether the async executors are running

Returns:

  • (Boolean) 


287
288
289
 
# File 'lib/good_job/adapter.rb', line 287

def async_started?
  @_async_started
end

#enqueue(active_job) ⇒ GoodJob::Job

Enqueues the ActiveJob job to be performed. For use by Rails; you should generally not call this directly.

Parameters:

  • active_job (ActiveJob::Base) —

    the job to be enqueued from #perform_later

Returns:



41
42
43
 
# File 'lib/good_job/adapter.rb', line 41

def enqueue(active_job)
  enqueue_at(active_job, nil)
end

#enqueue_after_transaction_commit? ⇒ Boolean

Defines if enqueueing this job from inside an Active Record transaction automatically defers the enqueue to after the transaction commit.

Returns:

  • (Boolean) 


47
48
49
 
# File 'lib/good_job/adapter.rb', line 47

def enqueue_after_transaction_commit?
  GoodJob.configuration.enqueue_after_transaction_commit
end

#enqueue_all(active_jobs) ⇒ Integer

Enqueues multiple ActiveJob instances at once

Parameters:

  • active_jobs (Array<ActiveJob::Base>) —

    jobs to be enqueued

Returns:

  • (Integer) —

    number of jobs that were successfully enqueued



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/good_job/adapter.rb', line 54

def enqueue_all(active_jobs)
  active_jobs = Array(active_jobs)
  return 0 if active_jobs.empty?

  # If there is a currently open Bulk in the current thread, direct the
  # jobs there to (eventually) be enqueued using enqueue_all
  return if GoodJob::Bulk.capture(active_jobs, queue_adapter: self)

  Rails.application.executor.wrap do
    current_time = Time.current
    jobs = active_jobs.map do |active_job|
      GoodJob::Job.build_for_enqueue(active_job).tap do |job|
        job.scheduled_at = current_time if job.scheduled_at == job.created_at
        job.created_at = current_time
        job.updated_at = current_time
      end
    end

    inline_jobs = []
    lock_strategy = GoodJob::Job.effective_lock_strategy
    tracker_registered = false
    lock_id = nil

    if execute_inline?
      @capsule.tracker.register
      tracker_registered = true
      lock_id = @capsule.tracker.id_for_lock
    end

    begin
      GoodJob::Job.transaction(requires_new: true, joinable: false) do
        column_names = GoodJob::Job.column_names
        job_attributes = jobs.map { |job| job.attributes.slice(*column_names) }
        results = GoodJob::Job.insert_all(job_attributes, returning: %w[id active_job_id]) # rubocop:disable Rails/SkipsModelValidations

        job_id_to_provider_job_id = results.to_h { |result| [result['active_job_id'], result['id']] }
        active_jobs.each do |active_job|
          active_job.provider_job_id = job_id_to_provider_job_id[active_job.job_id]
          active_job.successfully_enqueued = active_job.provider_job_id.present? if active_job.respond_to?(:successfully_enqueued=)
        end
        jobs.each do |job|
          job.instance_variable_set(:@new_record, false) if job_id_to_provider_job_id[job.active_job_id]
        end
        jobs = jobs.select(&:persisted?) # prune unpersisted jobs

        if execute_inline?
          inline_jobs = jobs.select { |job| job.scheduled_at.nil? || job.scheduled_at <= Time.current }
          if lock_strategy != :advisory && lock_id && inline_jobs.any?
            GoodJob::Job.where(id: inline_jobs.map(&:id)).update_all( # rubocop:disable Rails/SkipsModelValidations
              locked_by_id: lock_id, locked_at: current_time, lock_type: GoodJob::Job.lock_types[lock_strategy.to_s]
            )
            inline_jobs.each { |j| j.assign_attributes(locked_by_id: lock_id, locked_at: current_time, lock_type: lock_strategy) }
          end
          case lock_strategy
          when :advisory, :hybrid
            inline_jobs.each(&:advisory_lock!)
          end
        end
      end

      if inline_jobs.any?
        deferred = InlineBuffer.defer?
        InlineBuffer.perform_now_or_defer do
          until inline_jobs.empty?
            inline_job = inline_jobs.shift
            perform_inline(inline_job, notify: deferred ? send_notify?(inline_job) : false, already_claimed: lock_strategy != :advisory, advisory_unlock: lock_strategy != :skiplocked)
          end
        ensure
          inline_jobs.each(&:advisory_unlock)
          @capsule.tracker.unregister if tracker_registered
          tracker_registered = false
        end
      elsif tracker_registered
        @capsule.tracker.unregister
        tracker_registered = false
      end
    rescue StandardError
      if tracker_registered
        @capsule.tracker.unregister
        tracker_registered = false
      end
      raise
    end

    non_inline_jobs = if InlineBuffer.defer?
                        jobs - inline_jobs
                      else
                        jobs.reject(&:finished_at)
                      end
    if non_inline_jobs.any?
      job_id_to_active_jobs = active_jobs.index_by(&:job_id)
      non_inline_jobs.group_by(&:queue_name).each do |queue_name, jobs_by_queue|
        jobs_by_queue.group_by(&:scheduled_at).each do |scheduled_at, jobs_by_queue_and_scheduled_at|
          state = { queue_name: queue_name, count: jobs_by_queue_and_scheduled_at.size }
          state[:scheduled_at] = scheduled_at if scheduled_at

          executed_locally = execute_async? && @capsule&.create_thread(state)
          unless executed_locally
            state[:count] = job_id_to_active_jobs.values_at(*jobs_by_queue_and_scheduled_at.map(&:active_job_id)).count { |active_job| send_notify?(active_job) }
            Notifier.notify(state) unless state[:count].zero?
          end
        end
      end
    end
  end

  active_jobs.count(&:provider_job_id)
end

#enqueue_at(active_job, timestamp) ⇒ GoodJob::Job

Enqueues an ActiveJob job to be run at a specific time. For use by Rails; you should generally not call this directly.

Parameters:

  • active_job (ActiveJob::Base) —

    the job to be enqueued from #perform_later

  • timestamp (Integer, nil) —

    the epoch time to perform the job

Returns:



168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
 
# File 'lib/good_job/adapter.rb', line 168

def enqueue_at(active_job, timestamp)
  scheduled_at = timestamp ? Time.zone.at(timestamp) : nil

  # If there is a currently open Bulk in the current thread, direct the
  # job there to be enqueued using enqueue_all
  return if GoodJob::Bulk.capture(active_job, queue_adapter: self)

  Rails.application.executor.wrap do
    will_execute_inline = execute_inline? && (scheduled_at.nil? || scheduled_at <= Time.current)
    will_retry_inline = will_execute_inline && CurrentThread.job&.active_job_id == active_job.job_id && !CurrentThread.retry_now

    if will_retry_inline
      job = GoodJob::Job.enqueue(
        active_job,
        scheduled_at: scheduled_at
      )
    elsif will_execute_inline
      lock_strategy = GoodJob::Job.effective_lock_strategy
      @capsule.tracker.register
      lock_id = @capsule.tracker.id_for_lock
      tracker_registered = true
      begin
        job = case lock_strategy
              when :skiplocked
                GoodJob::Job.enqueue(active_job, scheduled_at: scheduled_at, lock_id: lock_id, lock_type: :skiplocked)
              when :hybrid
                GoodJob::Job.enqueue(active_job, scheduled_at: scheduled_at, lock_id: lock_id, lock_type: :hybrid, create_with_advisory_lock: true)
              else
                GoodJob::Job.enqueue(active_job, scheduled_at: scheduled_at, create_with_advisory_lock: true)
              end
        InlineBuffer.perform_now_or_defer do
          perform_inline(job, notify: send_notify?(active_job), already_claimed: lock_strategy != :advisory, advisory_unlock: lock_strategy != :skiplocked)
        ensure
          @capsule.tracker.unregister if tracker_registered
          tracker_registered = false
        end
      rescue StandardError
        if tracker_registered
          @capsule.tracker.unregister
          tracker_registered = false
        end
        raise
      end
    else
      job = GoodJob::Job.enqueue(
        active_job,
        scheduled_at: scheduled_at
      )

      if job
        executed_locally = execute_async? && @capsule&.create_thread(job.job_state)
        Notifier.notify(job.job_state) if !executed_locally && send_notify?(active_job)
      end
    end

    job
  end
end

#execute_async? ⇒ Boolean

Whether in :async execution mode.

Returns:

  • (Boolean) 


256
257
258
259
 
# File 'lib/good_job/adapter.rb', line 256

def execute_async?
  execution_mode == :async_all ||
    (execution_mode.in?([:async, :async_server]) && GoodJob.configuration.in_webserver?)
end

#execute_externally? ⇒ Boolean

Whether in :external execution mode.

Returns:

  • (Boolean) 


263
264
265
266
267
 
# File 'lib/good_job/adapter.rb', line 263

def execute_externally?
  execution_mode.nil? ||
    execution_mode == :external ||
    (execution_mode.in?([:async, :async_server]) && !GoodJob.configuration.in_webserver?)
end

#execute_inline? ⇒ Boolean

Whether in :inline execution mode.

Returns:

  • (Boolean) 


271
272
273
 
# File 'lib/good_job/adapter.rb', line 271

def execute_inline?
  execution_mode == :inline
end

#execution_mode ⇒ Symbol?

This adapter’s execution mode

Returns:

  • (Symbol, nil) 


250
251
252
 
# File 'lib/good_job/adapter.rb', line 250

def execution_mode
  @_execution_mode_override || GoodJob.configuration.execution_mode
end

#shutdown(timeout: NONE) ⇒ void

This method returns an undefined value.

Shut down the thread pool executors.

Parameters:

  • timeout (nil, Numeric, NONE) (defaults to: NONE) —

    Seconds to wait for active threads.

    • nil trigger a shutdown but not wait for it to complete.

    • -1 wait until the shutdown is complete.

    • 0 immediately shutdown and stop any threads.

    • A positive number will wait that many seconds before stopping any remaining active threads.



243
244
245
246
 
# File 'lib/good_job/adapter.rb', line 243

def shutdown(timeout: NONE)
  @capsule&.shutdown(timeout: timeout)
  @_async_started = false
end

#start_async ⇒ void

This method returns an undefined value.

Start async executors



277
278
279
280
281
282
283
 
# File 'lib/good_job/adapter.rb', line 277

def start_async
  return unless execute_async?

  @capsule.start
  @capsule.lower_thread_priority = true if GoodJob.configuration.lower_thread_priority.in?([true, nil])
  @_async_started = true
end

#stopping? ⇒ Boolean

Whether the current job execution thread is shutting down. This is intended to be called from within the context of a performing job. Jobs can call this while performing via ‘queue_adapter.stopping?` to support job continuations/iterations.

Returns:

  • (Boolean) 


232
233
234
 
# File 'lib/good_job/adapter.rb', line 232

def stopping?
  GoodJob.current_thread_shutting_down?
end