               multi.exec (multi_err, multi_res) =>
                 if multi_err
                   client.quit()
-                  return callback multi_err 
+                  return callback multi_err
                 if multi_res
                   client.quit()
+                  @redis.publish(@key('ENQUEUED'), "")
                   @statistics callback
                 else
                   reschedule callback

  #       console.log param1, param2
  #       callback()
  regist: (@handler) =>
    registered_workers.push "#{@fairy.id}|#{@name}"
    worker_id = uuid.v4()
    @redis.hset @key('WORKERS'), worker_id, "#{os.hostname()}|#{server_ip()}|#{process.pid}|#{Date.now()}"

    process.on 'uncaughtException', (err) =>
      if @_handler_callback
        console.log "Worker [#{worker_id.split('-')[0]}] registered for Task [#{@name}] will block its current processing group" 
        @_handler_callback {do: 'block', message: err.stack}
      else
        @_try_exit()

    process.on 'exit', => @redis.hdel @key('WORKERS'), worker_id
    @_poll()


  # ### Poll New Task

  # **Private** method. If any task presents in the `SOURCE` list, `lpop` from
  # `SOURCE` (`rpush`) into `QUEUED`. The `lpop` and `rpush` are protected by
  # a transaction.
  #
  # Since the task being popped and pushed should be known in prior of the
  # begin of the transaction (aka, the `multi` command), we need to get the
  # first task of the source list, and, watch the source list to prevent the
  # same task being taken by two different workers.
  #
  # If there's no pending tasks of the same group, then process the task
  # immediately.
  #
  # If there's no tasks in the `SOURCE` list, poll again after an interval of
  # `polling_interval` milliseconds.
  _poll: =>
    return @_try_exit() if exiting
    @redis.watch @key('SOURCE')
    @redis.lindex @key('SOURCE'), 0, (err, res) =>
      if res
        task = JSON.parse res
        @redis.multi()
        .lpop(@key('SOURCE'))
        .rpush("#{@key('QUEUED')}:#{task[1]}", res)
        .exec (multi_err, multi_res) =>
          return @_poll() unless multi_res and multi_res[1] is 1
          @_process task
      else
        @redis.unwatch()
        setTimeout @_poll, @polling_interval


  # ### Exit When All Queues are Cleaned Up

  # **Private** method. Wait if there're queues still working, or exit the
  # process immediately.
  _try_exit: =>
    registered_workers.splice registered_workers.indexOf "#{@fairy.id}|#{@name}", 1
    process.exit() unless registered_workers.length
    log_registered_workers()


  # ### Process Each Group's First Task

  # **Private** method. The real job is done by the passed in `handler` of
  # `regist`ered method, when the job is:
  #
  #   + **successed**, pop the finished job from the group queue, and:
  #     - continue process task of the same group if there's pending job(s) in
  #     the same `QUEUED` list, or
  #     - poll task from the `SOURCE` queue.
  #   + **failed**, then inspect the passed in argument, retry or block
  #   according to the `do` property of the error object.
  #
  # Calling the callback function is the responsibility of you. Otherwise
  # `Fairy` will stop dispatching tasks.
  _process: (task) =>

    @redis.hset @key('PROCESSING'), task[0], JSON.stringify([task..., start_time = Date.now()])

    # Before Processing the Task:
    #
    #   1. Keep start time of processing.
    #   2. Set `PROCESSING` hash in Redis.
    #   3. Allow `retry_limit` times of retries.
    processing = task[0]
    retry_count = @retry_limit
    errors = []

    @_handler_callback = handler_callback = (err, res) =>

      @_handler_callback = null

      # Error handling routine:
      #
      #   1. Keep the error message.
      #   2. According to specific error handling request, when:
      #     + `block`: block the group immediately.
      #     + `block-after-retry`: retry n times, block the group if still
      #     fails (blocking).
      #     + or: retry n times, skip the task if still fails (non-blocking).
      #     Set `retry_limit` to `0` to skip failed tasks immediately.
      if err
        errors.push err.message or null
        switch err.do
          when 'block'
            @redis.multi()
              .rpush(@key('FAILED'), JSON.stringify([task..., Date.now(), errors]))
              .hdel(@key('PROCESSING'), processing)
              .sadd(@key('BLOCKED'), task[1])
              .exec()
            return @_poll()
          when 'block-after-retry'
            return setTimeout call_handler, @retry_delay if retry_count--
            @redis.multi()
              .rpush(@key('FAILED'), JSON.stringify([task..., Date.now(), errors]))
              .hdel(@key('PROCESSING'), processing)
              .sadd(@key('BLOCKED'), task[1])
              .exec()
            return @_poll()
          else
            return setTimeout call_handler, @retry_delay if retry_count--
            @redis.multi()
              .rpush(@key('FAILED'), JSON.stringify([task..., Date.now(), errors]))
              .hdel(@key('PROCESSING'), processing)
              .exec()

      # Success handling routine:
      #
      #   1. Remove last task from processing hash.
      #   2. Update statistics hash:
      #     + total number of `finished` tasks;
      #     + total `pending` time;
      #   3. Track recent finished tasks in `RECENT` list.
      #   4. Track tasks take the longest processing time in `SLOWEST` sorted
      #   set.
      else
        finish_time  = Date.now()
        process_time = finish_time - start_time
        @redis.multi()
          .hdel(@key('PROCESSING'), processing)
          .hincrby(@key('STATISTICS'), 'FINISHED', 1)
          .hincrby(@key('STATISTICS'), 'TOTAL_PENDING_TIME', start_time - task[task.length - 1])
          .hincrby(@key('STATISTICS'), 'TOTAL_PROCESS_TIME', process_time)
          .lpush(@key('RECENT'), JSON.stringify([task..., finish_time]))
          .ltrim(@key('RECENT'), 0, @recent_size - 1)
          .zadd(@key('SLOWEST'), process_time, JSON.stringify([task..., start_time]))
          .zremrangebyrank(@key('SLOWEST'), 0, - @slowest_size - 1)
          .exec()

      @_continue_group task[1]

    do call_handler = =>
      @handler task[1...-1]..., (@_handler_callback = handler_callback)

  # ### Continue Process a Group

  # **Private** method. Upon successful execution of a task, or skipping a
  # failed task:
  #
  #   1. `lpop` the current task from `QUEUED` list.
  #   2. Check if there exists task in the same `QUEUED` list.
  #   3. Process if `YES`, or poll `SOURCE` if `NO`.
  #
  # Above commands are protected by a transaction to prevent multiple workers
  # processing a same task.
  _continue_group: (group) =>
    @redis.watch "#{@key('QUEUED')}:#{group}"
    @redis.lindex "#{@key('QUEUED')}:#{group}", 1, (err, res) =>
      if res
        task = JSON.parse res
        @redis.unwatch()
        @redis.lpop "#{@key('QUEUED')}:#{group}"
        return @_requeue_group group if exiting
        @_process task
      else
        @redis.multi()
        .lpop("#{@key('QUEUED')}:#{group}")
        .exec (multi_err, multi_res) =>
          return @_continue_group group unless multi_res   
          return @_try_exit() if exiting
          @_poll()
  

  # ### Requeue Tasks on Exit

  # **Private** method. Before exiting, requeue all tasks in the processing
  # `QUEUED` list back into `SOURCE` list to **prevent blocking** the group.
  #
  # To ensure the correct order of tasks, `lpush` tasks in the `QUEUED` list
  # in their reverse order. The `lrange` and `lpush` is protected by a
  # transaction for atomicity since other workers may still shifting tasks in
  # the `SOURCE` list into `QUEUED` list.
  #
  # When tasks are requeued successfully, `_try_exit` the process.
  _requeue_group: (group) =>
    @redis.watch "#{@key('QUEUED')}:#{group}"
    @redis.lrange "#{@key('QUEUED')}:#{group}", 0, -1, (err, res) =>
      @redis.multi()
      .lpush("#{@key('SOURCE')}", res.reverse()...)
      .del("#{@key('QUEUED')}:#{group}")
      .exec (multi_err, multi_res) =>
        return @_requeue_group group unless multi_res
        return @_try_exit()

  # ### Re-Schedule Failed and Blocked Tasks



# Log active workers while waiting for all workers to clean-up.
log_registered_workers = ->
  console.log "\nFairy is waiting for #{workers.length} workers to clean-up before exit:"
  for worker in workers
    console.log "  * Client Id: [#{worker.fairy.id}], Task: [#{worker.name}]"

workers = []

# Fairy will enter cleanup mode before exit when:
#
