quick kestrel zeromq port attempt

.gitignore

node_modules

index.coffee

Events = require 'events'
class Queue extends Events.EventEmitter
  # Simple queue that stores pending jobs, or available workers.  If a job is
  # added and there are workers, send it to one of the workers.  If a worker
  # becomes available, and there are pending jobs, send it a job immediately.
  constructor: ->
    @workers = []
    @jobs = []
    @quitting = false

  # Public: Sends a job to a worker.
  #
  # worker - The String ID for the worker.
  # job    - The String job contents.
  #
  # Emits ('ready', worker, job).
  # Returns nothing.
  perform: (worker, job) ->
    @emit 'ready', worker, job
    @quit() if @quitting

  # Public: Registers a workers as unavailable.
  #
  # worker - The String ID for the worker.
  #
  # Returns nothing.
  removeWorker: (worker) ->
    index = @workers.indexOf worker
    if index > -1
      delete @workers[index]

  # Public: Registers a worker as available.
  #
  # worker - The String ID for the worker.
  #
  # Returns nothing.
  addWorker: (worker) ->
    if job = @jobs.shift()
      @perform worker, job
    else if @workers.indexOf(worker) == -1
      @workers.push(worker)

  # Public: Prepares to process a job.
  #
  # job - The String job contents.
  #
  # Returns nothing.
  addJob: (job) ->
    return if @quitting

    if worker = @workers.shift()
      @perform worker, job
    else
      @jobs.push job

  quit: ->
    @quitting = true
    if @quitting and @jobs.length == 0
      @emit 'empty'

exports.create = (args...) ->
  new Queue args...

queue.coffee

EMPTY = ''

# sets up the queue
queue = require('./workers').create()

# sends the job to the worker through the zeromq router
queue.on 'ready', (worker, job) ->
  console.log queue.jobs.length, 'jobs / ', queue.workers.length, 'workers'
  router.send [worker, EMPTY, job]

# sets up a timer to add random jobs
delay = parseFloat(process.env.RATE or 1) * 1000
timer = setInterval(->
  queue.addJob randJob().toString()
  console.log queue.jobs.length, 'jobs / ', queue.workers.length, 'workers'
, delay)

# generates random jobs.
randJob = ->
  Math.round(Math.random() * 1000) / 1000

zmq = require 'zmq'
router = zmq.socket 'router'

# the router receives messages from workers:
router.on 'message', (id, _, msg) ->
  msg = msg.toString()
  switch msg
    # the worker is available
    when 'add' then queue.addWorker(id)
    
    # just kidding, it's unavailable!
    when 'remove'
      queue.removeWorker(id)
      router.send [id, EMPTY, 'ok']
    else
      console.log "wtf:"
      console.log data

router.bindSync 'tcp://127.0.0.1:5555'

process.on 'SIGQUIT', ->
  console.log 'closing...'
  queue.on 'empty', ->
    router.close()
    process.exit()
  clearInterval timer
  queue.quit()

queue_test.coffee

assert = require 'assert'
Events = require 'events'
Queue = require '..'

calls = 0
expectedWorker = null
expectedJob = null

events = new Events.EventEmitter

events.on 'ready', (worker, job) ->
  calls += 1
  assert.equal expectedWorker, worker
  assert.equal expectedJob, job

queue = Queue.create events
assert.equal 0, queue.workers.length
assert.equal 0, queue.jobs.length

# add worker
events.emit 'message', '1'
assert.deepEqual ['1'], queue.workers
assert.equal 0, queue.jobs.length

# add worker
events.emit 'message', '2'
assert.deepEqual ['1', '2'], queue.workers
assert.equal 0, queue.jobs.length

# add job
expectedWorker = '1'
expectedJob = 'a'
queue.addJob expectedJob
assert.deepEqual ['2'], queue.workers
assert.equal 0, queue.jobs.length

# add worker
events.emit 'message', '3'
assert.deepEqual ['2', '3'], queue.workers
assert.equal 0, queue.jobs.length

# add job
expectedWorker = '2'
expectedJob = 'a'
queue.addJob expectedJob
assert.deepEqual ['3'], queue.workers
assert.equal 0, queue.jobs.length

# add job
expectedWorker = '3'
expectedJob = 'a'
queue.addJob expectedJob
assert.deepEqual [], queue.workers
assert.equal 0, queue.jobs.length

# add job
queue.addJob 'a'
assert.deepEqual [], queue.workers
assert.deepEqual ['a'], queue.jobs

# add job
queue.addJob 'b'
assert.deepEqual [], queue.workers
assert.deepEqual ['a', 'b'], queue.jobs

# add worker
expectedWorker = '1'
expectedJob = 'a'
events.emit 'message', '1'
assert.deepEqual [], queue.workers
assert.deepEqual ['b'], queue.jobs

# add worker
expectedWorker = '2'
expectedJob = 'b'
events.emit 'message', '2'
assert.deepEqual [], queue.workers
assert.deepEqual [], queue.jobs

process.on 'exit', ->
  assert.equal 5, calls

worker.rb

class Worker
  attr_reader :identity
  attr_accessor :working

  # Really basic ZeroMQ worker that gets jobs through a REQ socket.  It first
  # sends a message to signal its availability.  The received message is the
  # job.  This is the "Simple Pirate Pattern" from the ZeroMQ Guide:
  #
  # http://zguide.zeromq.org/page:all#Basic-Reliable-Queuing-Simple-Pirate-Pattern
  #
  # context - A ZeroMQ Context.  There should be 1 per app.
  # id      - Optional socket identity.  Default: Process.pid
  #
  def initialize(context, id = nil)
    @context = context
    @identity = (id || Process.pid).to_s
    @working = true
    @socket = nil
    @poller = nil
    @poller_item = nil
    @on_connect = nil
    @on_exit = nil
    @on_job = nil
  end

  # Public: Sets a block to be called when creating the socket.
  #
  #     worker.on_connect do |context|
  #       context.connect :REQ, 'tcp://127.0.0.1:5555'
  #     end
  #
  # Yields a ZMQ::Context.
  # Returns nothing.
  def on_connect(&block)
    @on_connect = block
  end

  # Public: Sets a block to be called when the worker is shutting down.  This
  # gives the worker a chance to tell the server it is unavailable.
  #
  #     worker.on_exit do |socket|
  #       socket.send 'bye'
  #       socket.recv # wait for the server to acknowledge
  #     end
  #
  # Yields a ZMQ::Socket.
  # Returns nothing.
  def on_exit(&block)
    @on_exit = block
  end

  # Public: Sets a block to be called when a job has been given to the worker.
  #
  #     worker.on_job do |job|
  #       do_some_work(job)
  #     end
  #
  # Yields a String job.
  # Returns nothing.
  def on_job(&block)
    @on_job = block
  end

  # Public: Starts the work loop.
  #
  # start_msg - Optional String message to send to the socket to indicate
  #             availability of this worker.
  #
  # Returns nothing.
  def perform(start_msg = nil)
    reconnect start_msg

    loop do
      if socket = poll
        res = @on_job.call(socket.recv_nonblock)
        socket.send(res) if @working
      end

      if !@working && close_socket(@on_exit)
        puts "#{@identity}> CLOSING..."
        return
      end

      if @working && !socket
        reconnect start_msg
      end
    end
  end

  # Public: Gracefully prepares to shut this worker down.  This stops the 
  # #perform loop after the current job finishes.
  #
  # Returns nothing.
  def quit
    puts "#{@identity}> QUITTING..."
    @working = false
  end

  # Internal: Polls the socket and returns a readable socket.  If nothing comes
  # back, try reconnecting.
  #
  # Returns a ZMQ::Socket, or nil.
  def poll
    @poller.poll 5000
    @poller.readables.first
  rescue Errno::EINTR
  end

  # Internal: Closes the socket if created.
  #
  # on_exit - Optional Block to call before shutting down.  If set, but the
  #           is not readable, wait for the next loop in #perform before trying
  #           to close the socket.
  #
  # Returns true if the socket is closed, or false.
  def close_socket(on_exit = nil)
    return true if !@socket

    if on_exit
      if @socket.events != ZMQ::POLLOUT
        return false # socket is readable, wait for the job 
      end
      @on_exit.call @socket
    end

    @poller.remove(@poller_item) if @poller_item
    @socket.close if @socket
    true
  end

  # Internal: Reconnects the socket to the server.  Sets up a ZMQ::Poller to
  # do the work.
  #
  # msg - Optional String message to send upon connecting.
  #
  # Returns nothing.
  def reconnect(msg = nil)
    puts "#{@identity}> CONNECTING..."
    @poller ||= ZMQ::Poller.new
    close_socket
    @socket = @on_connect.call(@context)
    @socket.identity = @identity
    @poller_item = ZMQ::Pollitem(@socket, ZMQ::POLLIN)
    @poller.register @poller_item
    @socket.send msg if msg
  end
end

# Example usage of the worker
require 'rbczmq'

# setup a context
ctx = ZMQ::Context.new

# these are hardcoded commands for queue.coffee
add_cmd = 'add'
remove_cmd = 'remove'

# create 5 workers in separate threads
workers = []
threads = []
5.times do |i|
  id = "#{Process.pid}-#{i}"
  worker = Worker.new ctx, id

  # connect to the server
  worker.on_connect do |ctx|
    ctx.connect :REQ, 'tcp://127.0.0.1:5555'
  end

  # tell the server the worker is unavailable
  worker.on_exit do |socket|
    socket.send remove_cmd
    print "Removing... "
    puts socket.recv
  end

  # each job is just a sleep
  worker.on_job do |job|
    print "sleeping #{job}s..."
    sleep job.to_f
    puts 'done'
    add_cmd
  end

  workers << worker
  threads << Thread.new { worker.perform(add_cmd) }
end

# `kill -QUIT {pid}` to gracefully shut the workers down
trap :QUIT do
  workers.each &:quit
end

threads.each &:join

# if the threads return, the workers are done.  destroy the context and go
# home.
ctx.destroy

Very cool. I definitely would like to use/try zeromq along side kestrel someday. The benefit is we won't need dedicated queue server boxes, just sockets and workers.

I was thinking -- Why not just write something that posted tweets from the stream to some imaginary service? No need to fuck with proxies on gauges :) Brandon gave me some feedback that spawned some new ideas. I wrote a really dumb ZMQ Heartbeat as a Celluloid Actor: https://gist.github.com/1872953

Thus far, the pusher proxy "just works" which gives me confidence in zeromq, but I need to see where if fails and how before putting tracking data there. Totally loved reading through this. I like the idea of recreating stuff in a different technology so you can learn. You should definitely submit this to rubyconf brazil. :) Really sweet that you can just add/remove workers to the router using messages. I've been thinking it could be cool/heretical to give all our app instances the ability to talk to each other. I don't have a great use case for it, but it just seems like all processes should be able to talk to each other really easily, be they an app instance, a worker, or whatever.

Totally. I don't think rails processes will talk to rails processes… But I totally want our apps to have zeromq inboxes. We've talked about tying different apps together before (GitHub + Jobs, GitHub + Gauges, Gist + Gauges)… For instance… we're separating Github and Gist into separate apps. It'd be awesome if GitHub would just forward all user profile changes to a Gist inbox, that Gist can incorporate at its leisure.