class Sidekiq::Batch

Constants

BID_EXPIRE_TTL
VERSION

Attributes

bid[R]
callback_queue[R]
created_at[R]
description[R]

Public Class Methods

new(existing_bid = nil) click to toggle source
# File lib/sidekiq/batch.rb, line 17
def initialize(existing_bid = nil)
  @bid = existing_bid || SecureRandom.urlsafe_base64(10)
  @existing = !(!existing_bid || existing_bid.empty?)  # Basically existing_bid.present?
  @initialized = false
  @created_at = Time.now.utc.to_f
  @bidkey = "BID-" + @bid.to_s
  @ready_to_queue = []
end

Private Class Methods

cleanup_redis(bid) click to toggle source
# File lib/sidekiq/batch.rb, line 262
def cleanup_redis(bid)
  Sidekiq.logger.debug {"Cleaning redis of batch #{bid}"}
  Sidekiq.redis do |r|
    r.del(
      "BID-#{bid}",
      "BID-#{bid}-callbacks-complete",
      "BID-#{bid}-callbacks-success",
      "BID-#{bid}-failed",

      "BID-#{bid}-success",
      "BID-#{bid}-complete",
      "BID-#{bid}-jids",
    )
  end
end
enqueue_callbacks(event, bid) click to toggle source
# File lib/sidekiq/batch.rb, line 203
def enqueue_callbacks(event, bid)
  batch_key = "BID-#{bid}"
  callback_key = "#{batch_key}-callbacks-#{event}"
  already_processed, _, callbacks, queue, parent_bid, callback_batch = Sidekiq.redis do |r|
    r.multi do
      r.hget(batch_key, event)
      r.hset(batch_key, event, true)
      r.smembers(callback_key)
      r.hget(batch_key, "callback_queue")
      r.hget(batch_key, "parent_bid")
      r.hget(batch_key, "callback_batch")
    end
  end

  return if already_processed == 'true'

  queue ||= "default"
  parent_bid = !parent_bid || parent_bid.empty? ? nil : parent_bid    # Basically parent_bid.blank?
  callback_args = callbacks.reduce([]) do |memo, jcb|
    cb = Sidekiq.load_json(jcb)
    memo << [cb['callback'], event, cb['opts'], bid, parent_bid]
  end

  opts = {"bid" => bid, "event" => event}

  # Run callback batch finalize synchronously
  if callback_batch
    # Extract opts from cb_args or use current
    # Pass in stored event as callback finalize is processed on complete event
    cb_opts = callback_args.first&.at(2) || opts

    Sidekiq.logger.debug {"Run callback batch bid: #{bid} event: #{event} args: #{callback_args.inspect}"}
    # Finalize now
    finalizer = Sidekiq::Batch::Callback::Finalize.new
    status = Status.new bid
    finalizer.dispatch(status, cb_opts)

    return
  end

  Sidekiq.logger.debug {"Enqueue callback bid: #{bid} event: #{event} args: #{callback_args.inspect}"}

  if callback_args.empty?
    # Finalize now
    finalizer = Sidekiq::Batch::Callback::Finalize.new
    status = Status.new bid
    finalizer.dispatch(status, opts)
  else
    # Otherwise finalize in sub batch complete callback
    cb_batch = self.new
    cb_batch.callback_batch = true
    Sidekiq.logger.debug {"Adding callback batch: #{cb_batch.bid} for batch: #{bid}"}
    cb_batch.on(:complete, "Sidekiq::Batch::Callback::Finalize#dispatch", opts)
    cb_batch.jobs do
      push_callbacks callback_args, queue
    end
  end
end
process_failed_job(bid, jid) click to toggle source
# File lib/sidekiq/batch.rb, line 149
def process_failed_job(bid, jid)
  _, pending, failed, children, complete, parent_bid = Sidekiq.redis do |r|
    r.multi do
      r.sadd("BID-#{bid}-failed", jid)

      r.hincrby("BID-#{bid}", "pending", 0)
      r.scard("BID-#{bid}-failed")
      r.hincrby("BID-#{bid}", "children", 0)
      r.scard("BID-#{bid}-complete")
      r.hget("BID-#{bid}", "parent_bid")

      r.expire("BID-#{bid}-failed", BID_EXPIRE_TTL)
    end
  end

  # if the batch failed, and has a parent, update the parent to show one pending and failed job
  if parent_bid
    Sidekiq.redis do |r|
      r.multi do
        r.hincrby("BID-#{parent_bid}", "pending", 1)
        r.sadd("BID-#{parent_bid}-failed", jid)
        r.expire("BID-#{parent_bid}-failed", BID_EXPIRE_TTL)
      end
    end
  end

  if pending.to_i == failed.to_i && children == complete
    enqueue_callbacks(:complete, bid)
  end
end
process_successful_job(bid, jid) click to toggle source
# File lib/sidekiq/batch.rb, line 180
def process_successful_job(bid, jid)
  failed, pending, children, complete, success, total, parent_bid = Sidekiq.redis do |r|
    r.multi do
      r.scard("BID-#{bid}-failed")
      r.hincrby("BID-#{bid}", "pending", -1)
      r.hincrby("BID-#{bid}", "children", 0)
      r.scard("BID-#{bid}-complete")
      r.scard("BID-#{bid}-success")
      r.hget("BID-#{bid}", "total")
      r.hget("BID-#{bid}", "parent_bid")

      r.srem("BID-#{bid}-failed", jid)
      r.srem("BID-#{bid}-jids", jid)
      r.expire("BID-#{bid}", BID_EXPIRE_TTL)
    end
  end

  # if complete or successfull call complete callback (the complete callback may then call successful)
  if (pending.to_i == failed.to_i && children == complete) || (pending.to_i.zero? && children == success)
    enqueue_callbacks(:complete, bid)
  end
end
push_callbacks(args, queue) click to toggle source
# File lib/sidekiq/batch.rb, line 280
def push_callbacks args, queue
  Sidekiq::Client.push_bulk(
    'class' => Sidekiq::Batch::Callback::Worker,
    'args' => args,
    'queue' => queue
  ) unless args.empty?
end

Public Instance Methods

callback_batch=(callback_batch) click to toggle source
# File lib/sidekiq/batch.rb, line 36
def callback_batch=(callback_batch)
  @callback_batch = callback_batch
  persist_bid_attr('callback_batch', callback_batch)
end
callback_queue=(callback_queue) click to toggle source
# File lib/sidekiq/batch.rb, line 31
def callback_queue=(callback_queue)
  @callback_queue = callback_queue
  persist_bid_attr('callback_queue', callback_queue)
end
description=(description) click to toggle source
# File lib/sidekiq/batch.rb, line 26
def description=(description)
  @description = description
  persist_bid_attr('description', description)
end
increment_job_queue(jid) click to toggle source
# File lib/sidekiq/batch.rb, line 110
def increment_job_queue(jid)
  @ready_to_queue << jid
end
invalidate_all() click to toggle source
# File lib/sidekiq/batch.rb, line 114
def invalidate_all
  Sidekiq.redis do |r|
    r.setex("invalidated-bid-#{bid}", BID_EXPIRE_TTL, 1)
  end
end
jobs() { || ... } click to toggle source
# File lib/sidekiq/batch.rb, line 55
def jobs
  raise NoBlockGivenError unless block_given?

  bid_data, Thread.current[:bid_data] = Thread.current[:bid_data], []

  begin
    if !@existing && !@initialized
      parent_bid = Thread.current[:batch].bid if Thread.current[:batch]

      Sidekiq.redis do |r|
        r.multi do
          r.hset(@bidkey, "created_at", @created_at)
          r.hset(@bidkey, "parent_bid", parent_bid.to_s) if parent_bid
          r.expire(@bidkey, BID_EXPIRE_TTL)
        end
      end

      @initialized = true
    end

    @ready_to_queue = []

    begin
      parent = Thread.current[:batch]
      Thread.current[:batch] = self
      yield
    ensure
      Thread.current[:batch] = parent
    end

    return [] if @ready_to_queue.size == 0

    Sidekiq.redis do |r|
      r.multi do
        if parent_bid
          r.hincrby("BID-#{parent_bid}", "children", 1)
          r.hincrby("BID-#{parent_bid}", "total", @ready_to_queue.size)
          r.expire("BID-#{parent_bid}", BID_EXPIRE_TTL)
        end

        r.hincrby(@bidkey, "pending", @ready_to_queue.size)
        r.hincrby(@bidkey, "total", @ready_to_queue.size)
        r.expire(@bidkey, BID_EXPIRE_TTL)

        r.sadd(@bidkey + "-jids", @ready_to_queue)
        r.expire(@bidkey + "-jids", BID_EXPIRE_TTL)
      end
    end

    @ready_to_queue
  ensure
    Thread.current[:bid_data] = bid_data
  end
end
on(event, callback, options = {}) click to toggle source
# File lib/sidekiq/batch.rb, line 41
def on(event, callback, options = {})
  return unless %w(success complete).include?(event.to_s)
  callback_key = "#{@bidkey}-callbacks-#{event}"
  Sidekiq.redis do |r|
    r.multi do
      r.sadd(callback_key, JSON.unparse({
        callback: callback,
        opts: options
      }))
      r.expire(callback_key, BID_EXPIRE_TTL)
    end
  end
end
parent() click to toggle source
# File lib/sidekiq/batch.rb, line 126
def parent
  if parent_bid
    Sidekiq::Batch.new(parent_bid)
  end
end
parent_bid() click to toggle source
# File lib/sidekiq/batch.rb, line 120
def parent_bid
  Sidekiq.redis do |r|
    r.hget(@bidkey, "parent_bid")
  end
end
valid?(batch = self) click to toggle source
# File lib/sidekiq/batch.rb, line 132
def valid?(batch = self)
  valid = !Sidekiq.redis { |r| r.exists("invalidated-bid-#{batch.bid}") }
  batch.parent ? valid && valid?(batch.parent) : valid
end

Private Instance Methods

persist_bid_attr(attribute, value) click to toggle source
# File lib/sidekiq/batch.rb, line 139
def persist_bid_attr(attribute, value)
  Sidekiq.redis do |r|
    r.multi do
      r.hset(@bidkey, attribute, value)
      r.expire(@bidkey, BID_EXPIRE_TTL)
    end
  end
end