Permalink
Browse files

Change LockerBase#lock signature, tests

Change lock to take an options hash with :block and :timeout options.
Older style is still supported, but marked as deprecated in the docs.
  • Loading branch information...
1 parent 4a31422 commit ceb2ca070368ff479c651dcf828b0d88d777eaa3 @slyphon slyphon committed Aug 29, 2012
@@ -17,18 +17,8 @@ class ExclusiveLocker < LockerBase
# (see LockerBase#lock)
# obtain an exclusive lock.
#
- def lock(blocking=false)
- return true if synchronize { @locked }
- create_lock_path!(EXCLUSIVE_LOCK_PREFIX)
-
- if got_write_lock?
- synchronize { @locked = true }
- elsif blocking
- block_until_write_lock!
- else
- cleanup_lock_path!
- false
- end
+ def lock(opts={})
+ super
end
# (see LockerBase#assert!)
@@ -54,6 +44,21 @@ def acquirable?
end
private
+ def lock_with_opts_hash(opts)
+ create_lock_path!(EXCLUSIVE_LOCK_PREFIX)
+
+ block, timeout = opts.values_at(:block, :timeout)
+
+ if got_write_lock?
+ synchronize { @locked = true }
+ elsif block
+ block_until_write_lock!(:timeout => timeout)
+ else
+ cleanup_lock_path!
+ false
+ end
+ end
+
# the node that is next-lowest in sequence number to ours, the one we
# watch for updates to
def next_lowest_node
@@ -70,7 +75,7 @@ def got_write_lock?
end
alias got_lock? got_write_lock?
- def block_until_write_lock!
+ def block_until_write_lock!(opts={})
begin
path = "#{root_lock_path}/#{next_lowest_node}"
logger.debug { "#{self.class}##{__method__} path=#{path.inspect}" }
@@ -85,7 +90,7 @@ def block_until_write_lock!
logger.debug { "calling block_until_deleted" }
Thread.pass
- @node_deletion_watcher.block_until_deleted
+ @node_deletion_watcher.block_until_deleted(opts)
rescue WeAreTheLowestLockNumberException
ensure
logger.debug { "block_until_deleted returned" }
@@ -149,8 +149,21 @@ def unlock!
unlock
end
- # @param blocking [true,false] if true we block the caller until we can obtain
- # a lock on the resource
+ # @overload lock(blocking=false)
+ # @param blocking [true,false] if true we block the caller until we can
+ # obtain a lock on the resource
+ #
+ # @deprecated in favor of the options hash style
+ #
+ # @overload lock(opts={})
+ # @option opts [true,false] :block (false) if true we block the
+ # caller until we obtain a lock on the resource
+ #
+ # @option opts [Numeric] :timeout (nil) if given, the number of seconds
+ # we should wait for the lock to be acquired. Will raise
+ # LockWaitTimeoutError if we exceed the timeout.
+ #
+ # @since 1.7
#
# @return [true] if we're already obtained a shared lock, or if we were able to
# obtain the lock in non-blocking mode.
@@ -162,16 +175,27 @@ def unlock!
# @raise [InterruptedSession] raised when blocked waiting for a lock and
# the underlying client's session is interrupted.
#
- # @see ZK::Client::Unixisms#block_until_node_deleted more about possible execptions
- def lock(blocking=false)
- raise NotImplementedError
+ # @raise [LockWaitTimeoutError] if the given timeout is exceeded waiting
+ # for the lock to be acquired
+ #
+ # @see ZK::Client::Unixisms#block_until_node_deleted for more about possible execptions
+ def lock(opts={})
+ return true if @mutex.synchronize { @locked }
+
+ case opts
+ when TrueClass, FalseClass # old style boolean argument
+ opts = { :block => opts }
+ end
+
+ lock_with_opts_hash(opts)
end
- # (see #lock)
+ # delegates to {#lock}
+ #
# @deprecated the use of lock! is deprecated and may be removed or have
# its semantics changed in a future release
- def lock!(blocking=false)
- lock(blocking)
+ def lock!(opts={})
+ lock(opts)
end
# returns true if this locker is waiting to acquire lock
@@ -244,6 +268,10 @@ def assert!
end
private
+ def lock_with_opts_hash(opts={})
+ raise NotImplementedError
+ end
+
def synchronize
@mutex.synchronize { yield }
end
@@ -6,20 +6,8 @@ class SharedLocker < LockerBase
# (see LockerBase#lock)
# obtain a shared lock.
#
- def lock(blocking=false)
- return true if @locked
- create_lock_path!(SHARED_LOCK_PREFIX)
-
- if got_read_lock?
- @locked = true
- elsif blocking
- block_until_read_lock!
- else
- # we didn't get the lock, and we're not gonna wait around for it, so
- # clean up after ourselves
- cleanup_lock_path!
- false
- end
+ def lock(opts={})
+ super
end
# (see LockerBase#assert!)
@@ -93,24 +81,40 @@ def got_read_lock?
alias got_lock? got_read_lock?
private
- # TODO: make this generic, can either block or non-block
- def block_until_read_lock!
+ def lock_with_opts_hash(opts)
+ create_lock_path!(SHARED_LOCK_PREFIX)
+
+ block, timeout = opts.values_at(:block, :timeout)
+
+ if got_read_lock?
+ @locked = true
+ elsif block
+ block_until_read_lock!(:timeout => timeout)
+ else
+ # we didn't get the lock, and we're not gonna wait around for it, so
+ # clean up after ourselves
+ cleanup_lock_path!
+ false
+ end
+ end
+
+ def block_until_read_lock!(opts={})
begin
path = "#{root_lock_path}/#{next_lowest_write_lock_name}"
logger.debug { "SharedLocker#block_until_read_lock! path=#{path.inspect}" }
- synchronize do
+ @mutex.synchronize do
@node_deletion_watcher = NodeDeletionWatcher.new(zk, path)
@cond.broadcast
end
- @node_deletion_watcher.block_until_deleted
+ @node_deletion_watcher.block_until_deleted(opts)
rescue NoWriteLockFoundException
# next_lowest_write_lock_name may raise NoWriteLockFoundException,
# which means we should not block as we have the lock (there is nothing to wait for)
end
- synchronize { @locked = true }
+ @mutex.synchronize { @locked = true }
end
end # SharedLocker
end # Locker
@@ -126,7 +126,7 @@ def block_until_deleted(opts={})
logger.debug { "path #{path} was deleted" }
return true
when TIMED_OUT
- raise ZK::Exceptions::LockWaitTimeoutError
+ raise ZK::Exceptions::LockWaitTimeoutError, "timed out waiting for #{timeout.inspect} seconds for deletion of path: #{path.inspect}"
when INTERRUPTED
raise ZK::Exceptions::WakeUpException
when ZOO_EXPIRED_SESSION_STATE
@@ -148,14 +148,14 @@ def block_until_deleted(opts={})
def wait_for_result(timeout)
# do the deadline maths
time_to_stop = timeout ? (Time.now + timeout) : nil # slight time slippage between here
-
- until @result
+ #
+ until @result #
if timeout # and here
now = Time.now
if @result
return
- elsif (now > time_to_stop)
+ elsif (now >= time_to_stop)
@result = TIMED_OUT
return
end
@@ -82,13 +82,16 @@
end
describe 'blocking' do
+ let(:read_lock_path_template) { "/_zklocking/#{path}/#{ZK::Locker::SHARED_LOCK_PREFIX}" }
+
before do
zk.mkdir_p(root_lock_path)
+ @read_lock_path = zk.create(read_lock_path_template, '', :mode => :ephemeral_sequential)
+ @exc = nil
end
- it %[should block waiting for the lock] do
+ it %[should block waiting for the lock with old style lock semantics] do
ary = []
- read_lock_path = zk.create("/_zklocking/#{path}/read", '', :mode => :ephemeral_sequential)
locker.lock.should be_false
@@ -102,13 +105,62 @@
ary.should be_empty
locker.should_not be_locked
- zk.delete(read_lock_path)
+ zk.delete(@read_lock_path)
+
+ th.join(2).should == th
+
+ ary.length.should == 1
+ locker.should be_locked
+ end
+
+ it %[should block waiting for the lock with new style lock semantics] do
+ ary = []
+
+ locker.lock.should be_false
+
+ th = Thread.new do
+ locker.lock(:block => true)
+ ary << :locked
+ end
+
+ locker.wait_until_blocked(5)
+
+ ary.should be_empty
+ locker.should_not be_locked
+
+ zk.delete(@read_lock_path)
th.join(2).should == th
ary.length.should == 1
locker.should be_locked
end
+
+ it %[should time out waiting for the lock] do
+ ary = []
+
+ locker.lock.should be_false
+
+ th = Thread.new do
+ begin
+ locker.lock(:block => true, :timeout => 0.01)
+ ary << :locked
+ rescue Exception => e
+ @exc = e
+ end
+ end
+
+ locker.wait_until_blocked(5)
+
+ ary.should be_empty
+ locker.should_not be_locked
+
+ th.join(2).should == th
+
+ ary.should be_empty
+ @exc.should_not be_nil
+ @exc.should be_kind_of(ZK::Exceptions::LockWaitTimeoutError)
+ end
end # blocking
end # lock
end # ExclusiveLocker
Oops, something went wrong.

0 comments on commit ceb2ca0

Please sign in to comment.