Merge pull request #9 from amylsli/amy/support-cluster

[feature] support redis cluster

.gitignore

@@ -7,3 +7,7 @@
 /pkg/
 /spec/reports/
 /tmp/
+.idea
+/dump.rdb
+cluster-test/700*/appendonly.aof
+cluster-test/700*/nodes.conf

README.md

@@ -40,9 +40,27 @@ See [documentation](http://www.rubydoc.info/gems/master_lock) for advanced usage
 
 ## Development
 
-After checking out the repo, run `bin/setup` to install dependencies. Then, run `rake spec` to run the tests. You can also run `bin/console` for an interactive prompt that will allow you to experiment.
+After checking out the repo, run `bundle install` to install the gem dependencies. 
 
 To install this gem onto your local machine, run `bundle exec rake install`. To release a new version, update the version number in `version.rb`, and then run `bundle exec rake release`, which will create a git tag for the version, push git commits and tags, and push the `.gem` file to [rubygems.org](https://rubygems.org).
+### Testing
+If you do not have Redis set up, run `brew install redis`. This gives you access to `redis-server`.
+
+To set up the redis instance, run `redis-server` in the project level directory. The default config should be located at `/usr/local/etc/redis.conf`.
+
+To set up the redis cluster, copy your redis-server executable to `cluster-test/redis-server`. Open up 6 terminal tabs, and in every tab, start every instance:
+```
+cd cluster-test/7000
+../redis-server ./redis.conf
+```
+Assuming you have at least Redis 5, create your cluster by running the following:
+```
+redis-cli --cluster create 127.0.0.1:7000 127.0.0.1:7001 \
+127.0.0.1:7002 127.0.0.1:7003 127.0.0.1:7004 127.0.0.1:7005 \
+--cluster-replicas 1
+```
+
+Then, run `rake spec` to run the tests. 
 
 ## Contributing
 

cluster-test/7000/redis.conf

@@ -0,0 +1,5 @@
+port 7000
+cluster-enabled yes
+cluster-config-file nodes.conf
+cluster-node-timeout 5000
+appendonly yes

cluster-test/7001/redis.conf

@@ -0,0 +1,5 @@
+port 7001
+cluster-enabled yes
+cluster-config-file nodes.conf
+cluster-node-timeout 5000
+appendonly yes

cluster-test/7002/redis.conf

@@ -0,0 +1,5 @@
+port 7002
+cluster-enabled yes
+cluster-config-file nodes.conf
+cluster-node-timeout 5000
+appendonly yes

cluster-test/7003/redis.conf

@@ -0,0 +1,5 @@
+port 7003
+cluster-enabled yes
+cluster-config-file nodes.conf
+cluster-node-timeout 5000
+appendonly yes

cluster-test/7004/redis.conf

@@ -0,0 +1,5 @@
+port 7004
+cluster-enabled yes
+cluster-config-file nodes.conf
+cluster-node-timeout 5000
+appendonly yes

cluster-test/7005/redis.conf

@@ -0,0 +1,5 @@
+port 7005
+cluster-enabled yes
+cluster-config-file nodes.conf
+cluster-node-timeout 5000
+appendonly yes

lib/master_lock.rb

@@ -1,4 +1,5 @@
 require 'master_lock/version'
+require 'master_lock/backend'
 
 require 'logger'
 require 'socket'
@@ -32,10 +33,18 @@ class LockNotAcquiredError < StandardError; end
     :key_prefix,
     :redis,
     :sleep_time,
-    :ttl
+    :ttl,
+    :cluster
   )
 
   class << self
+    def backend
+      @backend ||= Backend.new
+    end
+
+    def backend=(backend)
+      @backend = backend
+    end
     # Obtain a mutex around a critical section of code. Only one thread on any
     # machine can execute the given block at a time. Returns the result of the
     # block.
@@ -55,105 +64,41 @@ class << self
     # @raise [NotStartedError] if called before {#start}
     # @raise [LockNotAcquiredError] if the lock cannot be acquired before the
     #   timeout
-    def synchronize(key, options = {})
-      check_configured
-      raise NotStartedError unless @registry
-
-      ttl = options[:ttl] || config.ttl
-      acquire_timeout = options[:acquire_timeout] || config.acquire_timeout
-      extend_interval = options[:extend_interval] || config.extend_interval
-
-      raise ArgumentError, "extend_interval cannot be negative" if extend_interval < 0
-      raise ArgumentError, "ttl must be greater extend_interval" if ttl <= extend_interval
-
-      if (options.include?(:if) && !options[:if]) ||
-          (options.include?(:unless) && options[:unless])
-        return yield
-      end
-
-      lock = RedisLock.new(
-        redis: config.redis,
-        key: key,
-        ttl: ttl,
-        owner: generate_owner
-      )
-      if !lock.acquire(timeout: acquire_timeout)
-        raise LockNotAcquiredError, key
-      end
-
-      registration =
-        @registry.register(lock, extend_interval)
-      logger.debug("Acquired lock #{key}")
-      begin
-        yield
-      ensure
-        @registry.unregister(registration)
-        if lock.release
-          logger.debug("Released lock #{key}")
-        else
-          logger.warn("Failed to release lock #{key}")
-        end
-      end
+    def synchronize(key, options = {}, &blk)
+      backend.synchronize(key, options, &blk)
     end
 
     # Starts the background thread to manage and extend currently held locks.
     # The thread remains alive for the lifetime of the process. This must be
     # called before any locks may be acquired.
     def start
-      @registry = Registry.new
-      Thread.new do
-        loop do
-          @registry.extend_locks
-          sleep(config.sleep_time)
-        end
-      end
+      backend.start
     end
 
     # Returns true if the registry has been started, otherwise false
     # @return [Boolean]
     def started?
-      !@registry.nil?
+      backend.started?
     end
 
     # Get the configured logger.
     #
     # @return [Logger]
     def logger
-      config.logger
+      backend.logger
     end
 
     # @return [Config] MasterLock configuration settings
     def config
-      if !defined?(@config)
-        @config = Config.new
-        @config.acquire_timeout = DEFAULT_ACQUIRE_TIMEOUT
-        @config.extend_interval = DEFAULT_EXTEND_INTERVAL
-        @config.hostname = Socket.gethostname
-        @config.logger = Logger.new(STDOUT)
-        @config.logger.progname = name
-        @config.key_prefix = DEFAULT_KEY_PREFIX
-        @config.sleep_time = DEFAULT_SLEEP_TIME
-        @config.ttl = DEFAULT_TTL
-      end
-      @config
+      backend.config
     end
 
     # Configure MasterLock using block syntax. Simply yields {#config} to the
     # block.
     #
     # @yield [Config] the configuration
-    def configure
-      yield config
-    end
-
-    private
-
-    def check_configured
-      raise UnconfiguredError, "redis must be configured" unless config.redis
-    end
-
-    def generate_owner
-      "#{config.hostname}:#{Process.pid}:#{Thread.current.object_id}"
+    def configure(&blk)
+      backend.configure(&blk)
     end
   end
 end

lib/master_lock/backend.rb

@@ -0,0 +1,125 @@
+module MasterLock
+  class Backend
+
+    # Obtain a mutex around a critical section of code. Only one thread on any
+    # machine can execute the given block at a time. Returns the result of the
+    # block.
+    #
+    # @param key [String] the unique identifier for the locked resource
+    # @option options [Fixnum] :ttl (60) the length of time in seconds before
+    #   the lock expires
+    # @option options [Fixnum] :acquire_timeout (5) the length of time to wait
+    #   to acquire the lock before timing out
+    # @option options [Fixnum] :extend_interval (15) the amount of time in
+    #   seconds that may pass before extending the lock
+    # @option options [Boolean] :if if this option is falsey, the block will be
+    #   executed without obtaining the lock
+    # @option options [Boolean] :unless if this option is truthy, the block will
+    #   be executed without obtaining the lock
+    # @raise [UnconfiguredError] if a required configuration variable is unset
+    # @raise [NotStartedError] if called before {#start}
+    # @raise [LockNotAcquiredError] if the lock cannot be acquired before the
+    #   timeout
+    def synchronize(key, options = {})
+      check_configured
+      raise NotStartedError unless @registry
+
+      ttl = options[:ttl] || config.ttl
+      acquire_timeout = options[:acquire_timeout] || config.acquire_timeout
+      extend_interval = options[:extend_interval] || config.extend_interval
+
+      raise ArgumentError, "extend_interval cannot be negative" if extend_interval < 0
+      raise ArgumentError, "ttl must be greater extend_interval" if ttl <= extend_interval
+
+      if (options.include?(:if) && !options[:if]) ||
+          (options.include?(:unless) && options[:unless])
+        return yield
+      end
+
+      lock = RedisLock.new(
+          redis: config.redis,
+          key: key,
+          ttl: ttl,
+          owner: generate_owner
+      )
+      if !lock.acquire(timeout: acquire_timeout)
+        raise LockNotAcquiredError, key
+      end
+
+      registration =
+          @registry.register(lock, extend_interval)
+      logger.debug("Acquired lock #{key}")
+      begin
+        yield
+      ensure
+        @registry.unregister(registration)
+        if lock.release
+          logger.debug("Released lock #{key}")
+        else
+          logger.warn("Failed to release lock #{key}")
+        end
+      end
+    end
+
+    # Starts the background thread to manage and extend currently held locks.
+    # The thread remains alive for the lifetime of the process. This must be
+    # called before any locks may be acquired.
+    def start
+      @registry = Registry.new
+      Thread.new do
+        loop do
+          @registry.extend_locks
+          sleep(config.sleep_time)
+        end
+      end
+    end
+
+    # Returns true if the registry has been started, otherwise false
+    # @return [Boolean]
+    def started?
+      !@registry.nil?
+    end
+
+    # Get the configured logger.
+    #
+    # @return [Logger]
+    def logger
+      config.logger
+    end
+
+    # @return [Config] MasterLock configuration settings
+    def config
+      if !defined?(@config)
+        @config = Config.new
+        @config.acquire_timeout = DEFAULT_ACQUIRE_TIMEOUT
+        @config.extend_interval = DEFAULT_EXTEND_INTERVAL
+        @config.hostname = Socket.gethostname
+        @config.logger = Logger.new(STDOUT)
+        @config.logger.progname = 'MasterLock'
+        @config.key_prefix = DEFAULT_KEY_PREFIX
+        @config.sleep_time = DEFAULT_SLEEP_TIME
+        @config.ttl = DEFAULT_TTL
+        @config.cluster = false
+      end
+      @config
+    end
+
+    # Configure MasterLock using block syntax. Simply yields {#config} to the
+    # block.
+    #
+    # @yield [Config] the configuration
+    def configure
+      yield config
+    end
+
+    private
+
+    def check_configured
+      raise UnconfiguredError, "redis must be configured" unless config.redis
+    end
+
+    def generate_owner
+      "#{config.hostname}:#{Process.pid}:#{Thread.current.object_id}"
+    end
+  end
+end

lib/master_lock/redis_lock.rb

@@ -95,7 +95,13 @@ def eval_script(script, script_hash, keys:, argv:)
     end
 
     def redis_key
-      "#{MasterLock.config.key_prefix}:#{key}"
+      # Key hash tags are a way to ensure multiple keys are allocated in the same hash slot.
+      # This allows our redis operations to work with clusters
+      if MasterLock.config.cluster
+        "{#{MasterLock.config.key_prefix}}:#{key}"
+      else
+        "#{MasterLock.config.key_prefix}:#{key}"
+      end
     end
   end
 end

lib/master_lock/version.rb

@@ -1,3 +1,3 @@
 module MasterLock
-  VERSION = "0.11.1"
+  VERSION = "0.12.0"
 end