Class: Consul::Mutex

Inherits:

Object

Object
Consul::Mutex

show all

Defined in:: lib/consul/mutex.rb

Overview

A Consul-mediated distributed mutex

Sometimes, you just want some code to run on only one machine in a cluster at any particular time. Perhaps you only need one copy running, and you'd like to have something ready to failover, or maybe you want to make sure you don't take down all your machines simultaneously for a code upgrade.

Either way, Consul::Mutex has got you covered.

Constant Summary collapse

ThreadExceptionError = Indicates something went wrong in the worker thread. Catch this exception if you specifically want to know that your worker thread lost its mind and errored out. The actual exception which caused the thread to terminate is available in #nested.

Class.new(Nesty::NestedStandardError)

ConsulError = Indicates some sort of problem communicating with Consul. Something has gone terribly, terribly wrong, and I need to tell you all about it.

Class.new(RuntimeError)

LostLockError = Indicates that the worker thread was terminated because we lost the distributed lock. You can't assume anything about what state anything is in that you haven't explicitly made true using ensure. In general, if you're getting this exception, something is doing terrible, terrible things to your Consul cluster.

Class.new(RuntimeError)

Instance Method Summary collapse

#initialize(key, opts = {}) ⇒ Mutex constructor
Create a new Consul-mediated distributed mutex.
#synchronize { ... } ⇒ Object
Run code under the protection of this mutex.

Constructor Details

#initialize(key, opts = {}) ⇒ `Mutex`

Create a new Consul-mediated distributed mutex.

Parameters:

key (String) —
the path (within the Consul KV store namespace, /v1/kv) which will be used as the lock key for all operations on this mutex. Every mutex created with the same key will exclude with all other mutexes created with the same key.
opts (Hash) (defaults to: {}) —
a customizable set of options

Options Hash (opts):

:value (String) — default: hostname —
the value to set on the lock key when we acquire the lock. This can be anything you like, but generally you'll want to set something that uniquely identifies who is currently holding the lock. The default, the local system's hostname, is generally a good choice.
:consul_url (String) — default: 'http://localhost:8500' —
where to connect to in order to talk to your local Consul cluster.

# File 'lib/consul/mutex.rb', line 100

def initialize(key, opts = {})
  @key        = key
  @value      = opts.fetch(:value, Socket.gethostname)
  @consul_url = URI(opts.fetch(:consul_url, 'http://localhost:8500'))
end

Instance Method Details

#synchronize { ... } ⇒ `Object`

Run code under the protection of this mutex.

This method works similarly to the Mutex#synchronize method in the Ruby stdlib. You pass it a block, and only one instance of the code in the block will be running at any given moment.

The big difference is that the blocks of code being mutually excluded can be running in separate processes, on separate machines. We are truly living in the future.

The slightly smaller difference is that the block of code you specify may not actually run, or it might be killed whilst running. You'll always get an exception raised if that occurs, but you'll need to be careful to clean up any state that your code sets using ensure blocks (or equivalent).

Your block of code is also run on a separate thread within the interpreter from the one in which #synchronize itself is called (in case that's important to you).

Yields:

your code will be run once we have acquired the lock which controls this mutex.

Raises:

(ArgumentError) —
if no block is passed to this method.
(RuntimeError) —
if some sort of internal logic error occurs (always a bug in this code, please report)
(SocketError) —
if a mysterious socket-related error occurs.
(ConsulError) —
if an error occurs talking to Consul. This is indicative of a problem with the Consul agent, or the network.
(ThreadExceptionError) —
if the worker thread exits with an exception during execution. This is indicative of a problem in your code.
(LostLockError) —
if the Consul lock is somehow disrupted while the worker thread is running. This should only happen if the Consul cluster has a serious problem of some sort, or someone fiddles with the lock key behind our backs. Find who is fiddling with the key, and break their fingers.

# File 'lib/consul/mutex.rb', line 149

def synchronize
  unless block_given?
    raise ArgumentError,
          "No block passed to #{self.inspect}#synchronize"
  end

  acquire_lock
  
  begin
    worker_thread  = Thread.new { yield }
    watcher_thread = Thread.new { hold_lock }

    tw = ThreadsWait.new(worker_thread, watcher_thread)
    finished_thread = tw.next_wait

    err = nil

    case finished_thread
      when worker_thread
        # Work completed successfully... excellent
        return worker_thread.value

      when watcher_thread
        # We lost the lock... fiddlesticks
        
        # May as well delete our session now, it's useless
        delete_session

        k = watcher_thread.value
        
        msg = if k.nil?
          "Lost lock, key deleted!"
        else
          if k.session.nil?
            "Lost lock, no active session!"
          else
            "Lost lock to session '#{k.session}'"
          end
        end

        raise LostLockError, msg
      else
        raise RuntimeError,
              "Mysterious return value from `ThreadsWait#next_wait: #{finished_thread.inspect}"
    end
  ensure
    watcher_thread.kill
    worker_thread.kill

    begin
      worker_thread.join
    rescue Exception => ex
      worker_thread = ex
    end

    begin
      watcher_thread.join
    rescue Exception => ex
      watcher_thread = ex
    end

    release_lock if @session_id

    if worker_thread.is_a?(Exception)
      raise ThreadExceptionError.new(
              "Worker thread raised exception",
              worker_thread
            )
    end
    
    if watcher_thread.is_a?(Exception)
      if watcher_thread.is_a?(ConsulError)
        raise watcher_thread
      else
        raise RuntimeError,
              "Watcher thread raised exception: " +
              "#{watcher_thread.message} " +
              "(#{watcher_thread.class})"
      end
    end
  end
end