aboutsummaryrefslogtreecommitdiffstats
path: root/activesupport/lib/active_support/concurrency/share_lock.rb
diff options
context:
space:
mode:
Diffstat (limited to 'activesupport/lib/active_support/concurrency/share_lock.rb')
-rw-r--r--activesupport/lib/active_support/concurrency/share_lock.rb88
1 files changed, 66 insertions, 22 deletions
diff --git a/activesupport/lib/active_support/concurrency/share_lock.rb b/activesupport/lib/active_support/concurrency/share_lock.rb
index ca48164c54..54244317e4 100644
--- a/activesupport/lib/active_support/concurrency/share_lock.rb
+++ b/activesupport/lib/active_support/concurrency/share_lock.rb
@@ -6,12 +6,6 @@ module ActiveSupport
# A share/exclusive lock, otherwise known as a read/write lock.
#
# https://en.wikipedia.org/wiki/Readers%E2%80%93writer_lock
- #--
- # Note that a pending Exclusive lock attempt does not block incoming
- # Share requests (i.e., we are "read-preferring"). That seems
- # consistent with the behavior of "loose" upgrades, but may be the
- # wrong choice otherwise: it nominally reduces the possibility of
- # deadlock by risking starvation instead.
class ShareLock
include MonitorMixin
@@ -48,17 +42,11 @@ module ActiveSupport
def start_exclusive(purpose: nil, compatible: [], no_wait: false)
synchronize do
unless @exclusive_thread == Thread.current
- if busy?(purpose)
+ if busy_for_exclusive?(purpose)
return false if no_wait
- loose_shares = @sharing.delete(Thread.current)
- @waiting[Thread.current] = compatible if loose_shares
-
- begin
- @cv.wait_while { busy?(purpose) }
- ensure
- @waiting.delete Thread.current
- @sharing[Thread.current] = loose_shares if loose_shares
+ yield_shares(purpose: purpose, compatible: compatible, block_share: true) do
+ @cv.wait_while { busy_for_exclusive?(purpose) }
end
end
@exclusive_thread = Thread.current
@@ -71,13 +59,19 @@ module ActiveSupport
# Relinquish the exclusive lock. Must only be called by the thread
# that called start_exclusive (and currently holds the lock).
- def stop_exclusive
+ def stop_exclusive(compatible: [])
synchronize do
raise "invalid unlock" if @exclusive_thread != Thread.current
@exclusive_depth -= 1
if @exclusive_depth == 0
@exclusive_thread = nil
+
+ if eligible_waiters?(compatible)
+ yield_shares(compatible: compatible, block_share: true) do
+ @cv.wait_while { @exclusive_thread || eligible_waiters?(compatible) }
+ end
+ end
@cv.broadcast
end
end
@@ -85,8 +79,16 @@ module ActiveSupport
def start_sharing
synchronize do
- if @exclusive_thread && @exclusive_thread != Thread.current
+ if @sharing[Thread.current] > 0 || @exclusive_thread == Thread.current
+ # We already hold a lock; nothing to wait for
+ elsif @waiting[Thread.current]
+ # We're nested inside a +yield_shares+ call: we'll resume as
+ # soon as there isn't an exclusive lock in our way
@cv.wait_while { @exclusive_thread }
+ else
+ # This is an initial / outermost share call: any outstanding
+ # requests for an exclusive lock get to go first
+ @cv.wait_while { busy_for_sharing?(false) }
end
@sharing[Thread.current] += 1
end
@@ -109,12 +111,12 @@ module ActiveSupport
# the block.
#
# See +start_exclusive+ for other options.
- def exclusive(purpose: nil, compatible: [], no_wait: false)
+ def exclusive(purpose: nil, compatible: [], after_compatible: [], no_wait: false)
if start_exclusive(purpose: purpose, compatible: compatible, no_wait: no_wait)
begin
yield
ensure
- stop_exclusive
+ stop_exclusive(compatible: after_compatible)
end
end
end
@@ -129,14 +131,56 @@ module ActiveSupport
end
end
+ # Temporarily give up all held Share locks while executing the
+ # supplied block, allowing any +compatible+ exclusive lock request
+ # to proceed.
+ def yield_shares(purpose: nil, compatible: [], block_share: false)
+ loose_shares = previous_wait = nil
+ synchronize do
+ if loose_shares = @sharing.delete(Thread.current)
+ if previous_wait = @waiting[Thread.current]
+ purpose = nil unless purpose == previous_wait[0]
+ compatible &= previous_wait[1]
+ end
+ compatible |= [false] unless block_share
+ @waiting[Thread.current] = [purpose, compatible]
+
+ @cv.broadcast
+ end
+ end
+
+ begin
+ yield
+ ensure
+ synchronize do
+ @cv.wait_while { @exclusive_thread && @exclusive_thread != Thread.current }
+
+ if previous_wait
+ @waiting[Thread.current] = previous_wait
+ else
+ @waiting.delete Thread.current
+ end
+ @sharing[Thread.current] = loose_shares if loose_shares
+ end
+ end
+ end
+
private
# Must be called within synchronize
- def busy?(purpose)
- (@exclusive_thread && @exclusive_thread != Thread.current) ||
- @waiting.any? { |k, v| k != Thread.current && !v.include?(purpose) } ||
+ def busy_for_exclusive?(purpose)
+ busy_for_sharing?(purpose) ||
@sharing.size > (@sharing[Thread.current] > 0 ? 1 : 0)
end
+
+ def busy_for_sharing?(purpose)
+ (@exclusive_thread && @exclusive_thread != Thread.current) ||
+ @waiting.any? { |t, (_, c)| t != Thread.current && !c.include?(purpose) }
+ end
+
+ def eligible_waiters?(compatible)
+ @waiting.any? { |t, (p, _)| compatible.include?(p) && @waiting.all? { |t2, (_, c2)| t == t2 || c2.include?(p) } }
+ end
end
end
end