slides

z, ?	toggle help (this)
space, →	next slide
shift-space, ←	previous slide
d	toggle debug mode
## <ret>	go to slide #
c, t	table of contents (vi)
f	toggle footer
r	reload slides
n	toggle notes
p	run preshow
P	toggle pause

loading presentation...

Jacob Burkhart

@beanstalksurf

take out the sleeps? .notes use the full work queue .notes Tim says look at app-engine pipeline .notes "eye" like bluebill or monit .notes FIND the celluloid loop (Tim can help) .notes Celluloid Logger

progression of sequential -> threaded -> multi-machine... easy -> hard .notes tim's list of interesting things https://gist.github.com/halorgium/3724a6c202f81f33c8c6

Ruby 2.0 GA

on

Engine Yard Cloud

How to Fail at Background Jobs

`jacobo.github.com/background_jobs_kaigi`

What do I know?

The simplest thing that can possibly work

••••••••

(0..50).each do |x|
  50.times{print [128000+x].pack "U*"}
end

Threads

••••••••

work = (0..50).to_a
worker = lambda do |x|
  50.times{print [128000+x].pack "U*"}
end

workers = work.map do |x|
  Thread.new do
    worker.call(x)
  end
end; "ok"
workers.map(&:join); "ok"

Thread Pool

••••••••

require 'thread'
work_q = Queue.new
(0..50).to_a.each{|x| work_q.push x }
workers = (0...4).map do
  Thread.new do
    begin
      while x = work_q.pop(true)
        50.times{print [128000+x].pack "U*"}
      end
    rescue ThreadError
    end
  end
end; "ok"
workers.map(&:join); "ok"

Redis (& Threads)

••••••••

require 'redis'
REDIS = Redis.connect
(0..50).to_a.each do |x|
  REDIS.rpush("jobs_q", x)
end
workers = (0...4).map do
  Thread.new do
    while(job = REDIS.lpop("jobs_q"))
      x = job.to_i
      50.times{print [128000+x].pack "U*"}
    end
  end
end
workers.map(&:join)

Celluloid

••••••••

require 'celluloid'
class Worker
  include Celluloid
  def work(x)
    50.times{print [128000+x].pack "U*"}
  end
end
worker_pool = Worker.pool(size: 4)
futures = (0..50).to_a.map do |x|
  worker_pool.future.work(x)
end
futures.map(&:value) #wait for all jobs
worker_pool.terminate

Futuroscope

••••••••

require 'forwardable'
require 'set'
require 'futuroscope/convenience'

Futuroscope.default_pool.max_workers = 4
(0..50).to_a.map do |x|
  future do
    50.times{print [128000+x].pack "U*"}
  end
end

`github.com/codegram/futuroscope`

DRB

••••••••

require 'drb'
server_pid = fork do
  DRb.start_service "druby://localhost:28371", (0..50).to_a
  DRb.thread.join
end
work_q = DRbObject.new nil, "druby://localhost:28371"
worker_pids = (0...4).map do |i|
  fork do
    #sleep(i) #else DRb::DRbConnError in ruby < 2.0
    while x = work_q.pop
      50.times{print [128000+x].pack "U*"}
    end
  end
end
worker_pids.each{ |pid| Process.wait(pid) }
Process.kill("KILL", server_pid)

STRESS the importance of killing your workers

DCell

••••••••

require 'dcell'
class Worker
  include Celluloid
  def work(x)
    50.times{sleep 0.1; print [128000+x].pack "U*"}
  end
end
worker_pids = (0...4).map do |i|
  fork do
    DCell.start id: "w#{i}", addr: "tcp://127.0.0.1:900#{i}"
    Worker.supervise_as "worker#{i}"
    sleep
  end
end
sleep 1 #wait for workers to start
DCell.start id: "main", addr: "tcp://127.0.0.1:9011"
futures = (0..50).map do |x|
  i = x % 4
  DCell::Node["w#{i}"]["worker#{i}"].future.work(x)
end
futures.map(&:value)
worker_pids.each{|pid| Process.kill("KILL", pid)}

The Story of a Bug

Engine Yard

Boot a Server

class Server < ActiveRecord::Base
  after_create do |s|
    ServerBootJob.async_boot_server(s.id)
  end
end

class ServerBootJob
  def boot_server(server_id)
    server = Server.find(server_id)
    ...
  end
end

Starling, Workling

class ServerBootJob < Workling::Base
  def boot_server(server_id)
    ...
  end
end
ServerBootJob.async_boot_server(server.id)

Workling::Remote.dispatcher =
  Workling::Remote::Runners::StarlingRunner.new

starling -f config/starling.yml
script/workling_client run

RabbitMQ, AMQP

connection = AMQP.connect(:host => '127.0.0.1')

channel  = AMQP::Channel.new(connection)
queue    = channel.queue("serverbooter")
exchange = channel.default_exchange

queue.subscribe do |payload|
  puts "Boot server: #{payload}"
end

exchange.publish JSON.encode(:server_id => 10),
                 :routing_key => "serverbooter"

`github.com/ruby-amqp/amqp`

`github.com/ruby-amqp/bunny`

ActiveRecord:: RecordNotFound

class Server < ActiveRecord::Base
  after_create do |s|
    ServerBootJob.async_boot_server(s.id)
  end
end

class ServerBootJob
  def boot_server(server_id)
    #sleep 1 ?
    server = Server.find(server_id)
    ...
  end
end

App Worker

Server.create

SQL INSERT

after_create

Job Enqueue

Job Dequeue

SQL SELECT

ERROR

SQL COMMIT

Hack ActiveRecord

class Server < ActiveRecord::Base
  after_create do |s|
    s.commit_callback do
      ServerBooter.async_boot_server(s.id)
    end
  end
end

...
def commit_callback
  self.connection.instance_eval do
    class << self
      alias commit_db_transaction_original commit_db_transaction
      ...

`github.com/brontes3d/commit_callback`

(rails 2.3 only)

Service Object

class ServerBooter
  def self.create_and_boot_server!(...)
    server = Server.create!(...)
    ServerBootJob.async_boot_server(server.id)
    server
  end
end

class ServerBootJob
  def boot_server(server_id)
    server = Server.find(server_id)
    ...
  end
end

after_commit

class Server < ActiveRecord::Base
  after_commit do |s|
    ServerBooter.async_boot_server(s.id)
  end
end

"If any exceptions are raised within one of these callbacks, they will be ignored so that they don’t interfere with the other callbacks" -- Rails Guide

`http://bitly.com/hlUiBc`

Asynchronous workers are like a service that hasn't been extracted

class Server < ActiveRecord::Base
  after_create do |s|
    s.remote_id = Provisioning::Client.post(
      "/boot_server", s.attributes_for_provisioning)
    s.save!
  end
end

3 Parts

Loop
Runner
Queue

3 Parts

Loop
Runner
Queue

class TrapLoop
  trap('TERM') { stop! }
  trap('INT')  { stop! }
  trap('SIGTERM') { stop! }

  def self.start(&block)
    @started = true
    @loop = true
    while(@loop) do
      yield
      safe_exit_point!
    end
  end

  def self.stop!
    @loop = false
  end

  def self.safe_exit_point!
    if @started && !@loop
      raise Interrupt
    end
  end

Unicorn

def worker_loop(worker)
  ...
  while sock = ready.shift
    if client = sock.kgio_tryaccept
      process_client(client)
      nr += 1
      worker.tick = Time.now.to_i
    end
    break if nr < 0
  end
  ...
end

def process_client(client)
  status, headers, body = 
    @app.call(env = @request.read(client))
  ...
end

Unicorn

def worker_loop(worker)
  ...
  while sock = ready.shift
    if client = sock.kgio_tryaccept
      process_client(client)
      nr += 1
      worker.tick = Time.now.to_i
    end
    break if nr < 0
  end
  ...
end

def process_client(client)
  status, headers, body =
    @app.call(env = @request.read(client))
  ...
end

Hack Unicorn

Unicorn::Worker.class_eval do
  alias :old_tick= :tick=
  def tick=(val)
    if(job = REDIS.lpop("jobs_q"))
      sleep 2
      x = job.to_i
      50.times{print [128000+x].pack "U*"}
    end
    self.old_tick = val
  end
end

EventMachine

void EventMachine_t::Run()
  //Epoll and Kqueue stuff..
  ...

  while (true) {
    _UpdateTime();
    _RunTimers();

    _AddNewDescriptors();
    _ModifyDescriptors();

    _RunOnce();
    if (bTerminateSignalReceived)
      break;
  }
}

github.com/eventmachine/eventmachine/blob/master/ext/em.cpp

Thin

EM.next_tick{ ... }

Resque

def work(interval = 5, &block)
  loop do
    run_hook :before_fork, job

    if job = reserve
      if @child = fork
        procline "Forked #{@child} at #{Time.now.to_i}"
        Process.wait
      else
        procline "Processing #{job.queue} since #{Time.now.to_i}"
        perform(job, &block)
        exit! unless @cant_fork
      end
    end
  end

github.com/defunkt/resque/blob/master/lib/resque/worker.rb

EM-Resque ?

`github.com/SponsorPay/em-resque`

OR...

worker = Resque::Worker.new("*")
EM.add_periodic_timer( 0.1 ) do
  if job = worker.reserve
    worker.perform(job)
  end
end

Sucker Punch

class Worker
  include SuckerPunch::Worker
  def perform(job)
    ...
  end
end

`github.com/brandonhilkert/sucker_punch`

SuckerPunch::Worker = Celluloid

Sidekiq

`sidekiq.org`

Reel

`github.com/celluloid/reel`

DCell

`github.com/celluloid/dcell`

Torquebox

`torquebox.org`

Failing at Loops

Lousy Hooks

after_create do |server|
  Thread.new{ server.boot! }
end

after_create do |server|
  EM.next_tick{ server.boot! }
end

after_create do |server|
  Rack.after_request{ server.boot! }
end

Cron jobs are hard

Let me write ruby

# (- installed on Fri May 24 16:47:05 2013)
# (Cron version V5.0 -- $Id: crontab.c,v 1.12 2004/01/23 18:56:42 vixie Exp $)
10 * * * * cd /data/app/current && RAILS_ENV=production script/periodic_thing

`blog.engineyard.com/2013/cron-jobs`

`github.com/bvandenbos/resque-scheduler`

resque is already in a loop, why can't it check for work and run it

WANTED

hooks!

Unicorn/Rack after-request
Resque idle-tick
Rails 4 Q Integration

3 Parts

Loop
Runner
Queue

God

5.times do |n|
  God.watch do |w|
    w.name     = "resque-#{num}"
    w.group    = 'resque'
    w.interval = 30.seconds
    w.log      = "#{app_root}/log/worker.#{num}.log"
    w.dir      = app_root
    w.env      = {
      "GOD_WATCH"   => w.name,
      "QUEUE"       => '*'
    }
    w.start    = "bundle exec rake --trace resque:work"
  ...

`godrb.com`

Daemons

require 'daemons'

options = {
  :app_name => "worker",
  :log_output => true,
  :backtrace => true,
  :dir_mode => :normal,
  :dir => File.expand_path('../../tmp/pids',  __FILE__),
  :log_dir => File.expand_path('../../log',  __FILE__),
  :multiple => true,
  :monitor => true
}

Daemons.run(File.expand_path('../worker',  __FILE__), options)

`daemons.rubyforge.org`

Monit

(mess of bash)

if [[ ! -f "$pidfile" ]]
then
  cmd="/usr/bin/env \$V \$VV 
    APP_ROOT=\${application_root} ${RAKE:-'rake'} -f 
    \${application_root}/Rakefile resque:work"

`mmonit.com/monit`

Bluepill

Bluepill.application("#{app_name}") do |app|
  app.working_dir = "/var/apps/#{app_name}/current"
  worker_count.times do |i|
    app.process("resque-#{i}") do |process|
      process.start_command = "bundle exec rake resque:work"
      process.pid_file = "/var/apps/canvas/shared/pids/#{app_name}-resque-#{i}.pid"
      process.stop_command = "kill -QUIT {{PID}}"
      process.daemonize = true
      process.monitor_children do |child_process|
        child_process.stop_command = "kill -9 {{PID}}"
      end
    end
  end
end

`github.com/arya/bluepill`

Failing at Runners

Deploying

Startup New Worker

Wait to Finish Current Job

Shutdown

Timeout?

Kill?

Do Work

Zombie Workers

Do not respond to kill

Come back from the dead and eat your jobs

Untracked Workers

A worker without a pidfile is not restarted

Idle Workers

Lose their connections

WANTED

Better Tools!

"Web Job Server" concept
Monitoring of the dead
Check "Code Version" before job execution

3 Parts

Loop
Runner
Queue

Queue Basics

`rpush("jobs", job_data)`

`job_data = lpop("jobs")`

DIY Redis Queue

REDIS = Redis.connect(:url => "redis://#{redishost}")

class ServerBootJob
  def self.enq_job(server_id)
    REDIS.rpush("server_boot_jobs", server_id)
  end

  def self.process_jobs!
    while(server_id = REDIS.lpop("server_boot_jobs"))
      if server = Server.find_by_id(server_id)
        server.boot!
      end
    end
  end
end

Resque Redis

queues of work
a list of queues
a list of workers
a list of jobs in progress
a list of failed jobs

Failing at Queues

Re-Enqueue

class ServerSetupJob

  def self.perform(server_id)
    server = Server.find(server_id)
    if server.still_booting?
      Resque.enqueue ServerSetupJob, server_id
    else
      #continue...
    end
  end

end

Truly Failing at Background Jobs

Job Dependencies

Job Race Conditions

1 ServerBoot Job

1 ServerTerminate Job

Resque Plugins

class ServerBooting
  extend Resque::Plugins::JobTracking

  def self.track(server_id, opts)
    s = Server.find(server_id)
    ["Account:#{s.account_id}",
     "Environment:#{s.environment_id}"]
     "Server:#{server_id}"]
  end

  def self.perform(server_id, opts)
    #do it
  end
end

`github.com/engineyard/resque-job-tracking`

225 is a lot of
Resque Plugins

`rubygems.org/search?query=resque`

WANTED

Better Tools!

Run-models
Inter-Worker Communication
?

@cellubeard

`github.com/halorgium/celluloid-coordinator`

"Best Practices"

Idempotence (retriable jobs)
Redis locks (with expire)
Resque.inline = true
Error Tracking (rollbar.com)

"Best Practices" cont.

Model jobs as ActiveRecord objects (persist to DB)
Pass Request IDs
Be Paranoid

You Are not Alone

We can do better

TODO

Contribute to Resque github.com/resque
Contribute to Celluloid github.com/celluloid

Jacob Burkhart

Ruby 2.0 GA

on

Engine Yard Cloud

How to Fail at Background Jobs

What do I know?

The simplest thing that can possibly work

Threads

Thread Pool

Redis (& Threads)

Celluloid

Futuroscope

DRB

DCell

The Story of a Bug

Engine Yard

Boot a Server

Starling, Workling

RabbitMQ, AMQP

ActiveRecord:: RecordNotFound

Hack ActiveRecord

Service Object

after_commit

See Also: http://bitly.com/IqdwGP

See Also: http://bitly.com/OLij33

after_commit

"If any exceptions are raised within one of these callbacks, they will be ignored so that they don’t interfere with the other callbacks" -- Rails Guide

Asynchronous workers are like a service that hasn't been extracted

3 Parts

3 Parts

Unicorn

Unicorn

Hack Unicorn

EventMachine

Thin

Resque

EM-Resque ?

OR...

Sucker Punch

Sidekiq

Reel

DCell

Torquebox

Failing at Loops

Lousy Hooks

Cron jobs are hard

Let me write ruby

WANTED

hooks!

3 Parts

God

Daemons

Monit

(mess of bash)

Bluepill

Failing at Runners

Deploying

Zombie Workers

Do not respond to kill

Come back from the dead and eat your jobs

Untracked Workers

A worker without a pidfile is not restarted

Idle Workers

Lose their connections

WANTED

Better Tools!

3 Parts

Queue Basics

rpush("jobs", job_data)

job_data = lpop("jobs")

DIY Redis Queue

Resque Redis

Failing at Queues

Re-Enqueue

Truly Failing at Background Jobs

Job Dependencies

Job Race Conditions

1 ServerBoot Job

1 ServerTerminate Job

Resque Plugins

See Also: `http://bitly.com/IqdwGP`

See Also: `http://bitly.com/OLij33`

`rpush("jobs", job_data)`

`job_data = lpop("jobs")`

225 is a lot of
Resque Plugins