Skip to content

EventDeliveryModel

Jump to bottom Edit New page
Jonathan D. Simms edited this page May 1, 2012 · 7 revisions

There are two choices for how you'd like your events delivered to you when using ZK, and you specify that choice using the :thread option to the callback, with the values :single and :per_callback.

Single Event Thread

This is the current default, and is how ZK has operated since before 0.8. Events are dispatched to callbacks using a single thread, in the order they come in, to the callbacks in the order they were registered.

require 'zk'

include ZookeeperConstants

logger = Logger.new($stderr).tap { |l| l.level = Logger::DEBUG }

zk = ZK.new

# a hack to inject events directly into to the event queue
def zk.deliver(path, type)
  event_handler.process(ZK::Event.new(:path => path, :type => type))
end

zk.register('/foo') do |event|
  logger.debug "FIRST! /foo callback #{event.event_name}"
end

zk.register('/foo') do |event|
  logger.debug "second /foo callback, #{event.event_name}"
end

zk.register('/bar') do |event|
  logger.debug "/bar callback, #{event.event_name}"
end

# simulates the event delivery thread
Thread.new do
  zk.deliver('/foo', ZOO_CREATED_EVENT)
  zk.deliver('/foo', ZOO_DELETED_EVENT)
  zk.deliver('/bar', ZOO_CREATED_EVENT)
  zk.deliver('/bar', ZOO_CHANGED_EVENT)
end

sleep(2)
produces the output:

DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT
DEBUG -- : second /foo callback, ZOO_CREATED_EVENT
DEBUG -- : FIRST! /foo callback ZOO_DELETED_EVENT
DEBUG -- : second /foo callback, ZOO_DELETED_EVENT
DEBUG -- : /bar callback, ZOO_CREATED_EVENT
DEBUG -- : /bar callback, ZOO_CHANGED_EVENT

This is how we would expect things to work, even in the case where there are pauses due to processing in one of the callbacks, only one callback is executed at a time, and events are delivered in order. This is fine for most simple applications and for initial development work, as it provides the most predictable execution.

Thread per Callback

In this model, each callback you register is given its own thread and queue. Events are delivered to the callback's queue, and the thread takes care of calling the callback with them one at a time. This model allows us to increase the parallelism of the app while also allowing us to guarantee:

  • Only one thread will be executing the callback at a time
  • Events will be delivered to the callback in the order they were received
  • A callback blocking will only affect the delivery of the next event to that callback.

The first two are also true for the single thread model, the last one is an advantage of the thread-per-callback model. The following difference is unique to the thread-per-callback-model, though:

  • The same event may be delivered at different times system-wide.

Which makes sense in a multi-threaded system.

A small example. Here, we're going to deliver 10 events on '/foo' and '/bar', and each callback block is going to add a small random sleep before returning (simulating some kind of "work performed"). You'll see in the output afterwards that while the events were delivered at different times to the different callbacks, the order to all callbacks was the same

require 'zk'

include ZookeeperConstants

logger = Logger.new($stderr).tap { |l| l.level = Logger::DEBUG }

zk = ZK.new('localhost:2181', :thread => :per_callback)

# a hack to inject events directly into to the event queue
def zk.deliver(path, type, context)
  event_handler.process(ZK::Event.new(:path => path, :type => type, :context => context))
end

zk.register('/foo') do |event|
  logger.debug "FIRST! /foo callback #{event.event_name}, #{event.context}"
  sleep(rand(10) * 0.001)
end

zk.register('/foo') do |event|
  logger.debug "second /foo callback, #{event.event_name}, #{event.context}"
  sleep(rand(10) * 0.001)
end

zk.register('/bar') do |event|
  logger.debug "/bar callback, #{event.event_name}, #{event.context}"
  sleep(rand(10) * 0.01)
end

# make this repeatable
srand(1)

events = [ZOO_CREATED_EVENT, ZOO_CHANGED_EVENT, ZOO_CHILD_EVENT, ZOO_CHANGED_EVENT]

# simulates the event delivery thread
Thread.new do
  10.times do |n|
    zk.deliver('/foo', events.sample, n)
    zk.deliver('/bar', events.sample, n)
  end
end

sleep(2)

Produces the output:

[01:17:41.709086] DEBUG -- : FIRST! /foo callback ZOO_CHANGED_EVENT, 0
[01:17:41.709221] DEBUG -- : second /foo callback, ZOO_CHANGED_EVENT, 0
[01:17:41.709293] DEBUG -- : /bar callback, ZOO_CHILD_EVENT, 0
[01:17:41.719466] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 1
[01:17:41.719633] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 1
[01:17:41.726058] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 2
[01:17:41.727401] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 3
[01:17:41.727517] DEBUG -- : second /foo callback, ZOO_CHANGED_EVENT, 4
[01:17:41.728783] DEBUG -- : second /foo callback, ZOO_CHANGED_EVENT, 5
[01:17:41.728884] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 2
[01:17:41.738058] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 6
[01:17:41.738199] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 3
[01:17:41.741572] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 7
[01:17:41.748336] DEBUG -- : FIRST! /foo callback ZOO_CHANGED_EVENT, 4
[01:17:41.750703] DEBUG -- : second /foo callback, ZOO_CHANGED_EVENT, 8
[01:17:41.754165] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 9
[01:17:41.756349] DEBUG -- : FIRST! /foo callback ZOO_CHANGED_EVENT, 5
[01:17:41.762182] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 6
[01:17:41.763585] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 7
[01:17:41.773710] DEBUG -- : FIRST! /foo callback ZOO_CHANGED_EVENT, 8
[01:17:41.777252] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 9
[01:17:41.780594] DEBUG -- : /bar callback, ZOO_CHANGED_EVENT, 1
[01:17:41.861766] DEBUG -- : /bar callback, ZOO_CREATED_EVENT, 2
[01:17:41.872999] DEBUG -- : /bar callback, ZOO_CHANGED_EVENT, 3
[01:17:41.913864] DEBUG -- : /bar callback, ZOO_CHILD_EVENT, 4
[01:17:41.913972] DEBUG -- : /bar callback, ZOO_CHILD_EVENT, 5
[01:17:41.944432] DEBUG -- : /bar callback, ZOO_CHANGED_EVENT, 6
[01:17:42.035611] DEBUG -- : /bar callback, ZOO_CHILD_EVENT, 7
[01:17:42.055915] DEBUG -- : /bar callback, ZOO_CHILD_EVENT, 8
[01:17:42.056029] DEBUG -- : /bar callback, ZOO_CREATED_EVENT, 9

So the '/bar' callback has a sleep multiplier that is one order-of-magnatude larger than the '/foo' callbacks, and we can see that it processes the first event and sleeps, but doesn't block the delivery of events to other callbacks. You can also see that both the first and second '/foo' callbacks receive their events in the same order (denoted by the 'context' number at the EOL).

Running that same code with :thread => :single we see

D, [01:25:00.288553] DEBUG -- : FIRST! /foo callback ZOO_CHANGED_EVENT, 0
D, [01:25:00.298346] DEBUG -- : second /foo callback, ZOO_CHANGED_EVENT, 0
D, [01:25:00.308591] DEBUG -- : /bar callback, ZOO_CHILD_EVENT, 0
D, [01:25:00.379793] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 1
D, [01:25:00.386812] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 1
D, [01:25:00.396146] DEBUG -- : /bar callback, ZOO_CHANGED_EVENT, 1
D, [01:25:00.407168] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 2
D, [01:25:00.407333] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 2
D, [01:25:00.408621] DEBUG -- : /bar callback, ZOO_CREATED_EVENT, 2
D, [01:25:00.489797] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 3
D, [01:25:00.498985] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 3
D, [01:25:00.502502] DEBUG -- : /bar callback, ZOO_CHANGED_EVENT, 3
D, [01:25:00.593444] DEBUG -- : FIRST! /foo callback ZOO_CHANGED_EVENT, 4
D, [01:25:00.602622] DEBUG -- : second /foo callback, ZOO_CHANGED_EVENT, 4
D, [01:25:00.610623] DEBUG -- : /bar callback, ZOO_CHILD_EVENT, 4
D, [01:25:00.641788] DEBUG -- : FIRST! /foo callback ZOO_CHANGED_EVENT, 5
D, [01:25:00.648270] DEBUG -- : second /foo callback, ZOO_CHANGED_EVENT, 5
D, [01:25:00.653985] DEBUG -- : /bar callback, ZOO_CHILD_EVENT, 5
D, [01:25:00.664595] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 6
D, [01:25:00.674749] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 6
D, [01:25:00.678275] DEBUG -- : /bar callback, ZOO_CHANGED_EVENT, 6
D, [01:25:00.719443] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 7
D, [01:25:00.728667] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 7
D, [01:25:00.729917] DEBUG -- : /bar callback, ZOO_CHILD_EVENT, 7
D, [01:25:00.770318] DEBUG -- : FIRST! /foo callback ZOO_CHANGED_EVENT, 8
D, [01:25:00.770449] DEBUG -- : second /foo callback, ZOO_CHANGED_EVENT, 8
D, [01:25:00.774027] DEBUG -- : /bar callback, ZOO_CHILD_EVENT, 8
D, [01:25:00.864362] DEBUG -- : FIRST! /foo callback ZOO_CREATED_EVENT, 9
D, [01:25:00.866898] DEBUG -- : second /foo callback, ZOO_CREATED_EVENT, 9
D, [01:25:00.867004] DEBUG -- : /bar callback, ZOO_CREATED_EVENT, 9

All progress is gated by the slow '/bar' callback blocking the (only) delivery thread.

Add a custom footer
Add a custom sidebar

Clone this wiki locally