Merge pull request #381 from alexdowad/reentrant_rw_lock

New ReentrantReadWriteLock class

Author: jdantonio

ext/com/concurrent_ruby/ext/JavaAtomicFixnumLibrary.java

@@ -13,6 +13,7 @@
 import org.jruby.runtime.ThreadContext;
 import org.jruby.runtime.builtin.IRubyObject;
 import org.jruby.runtime.load.Library;
+import org.jruby.runtime.Block;
 
 public class JavaAtomicFixnumLibrary implements Library {
 
@@ -88,6 +89,18 @@ public IRubyObject compareAndSet(ThreadContext context, IRubyObject expect, IRub
             return getRuntime().newBoolean(atomicLong.compareAndSet(rubyFixnumToLong(expect), rubyFixnumToLong(update)));
         }
 
+        @JRubyMethod
+        public IRubyObject update(ThreadContext context, Block block) {
+            for (;;) {
+                long _oldValue       = atomicLong.get();
+                IRubyObject oldValue = getRuntime().newFixnum(_oldValue);
+                IRubyObject newValue = block.yield(context, oldValue);
+                if (atomicLong.compareAndSet(_oldValue, rubyFixnumToLong(newValue))) {
+                    return newValue;
+                }
+            }
+        }
+
         private long rubyFixnumToLong(IRubyObject value) {
             if (value instanceof RubyFixnum) {
                 RubyFixnum fixNum = (RubyFixnum) value;

ext/concurrent/atomic_fixnum.c

@@ -60,3 +60,14 @@ VALUE method_atomic_fixnum_compare_and_set(VALUE self, VALUE rb_expect, VALUE rb
   Check_Type(rb_update, T_FIXNUM);
   return ir_compare_and_set(self, rb_expect, rb_update);
 }
+
+VALUE method_atomic_fixnum_update(VALUE self) {
+  VALUE old_value, new_value;
+  for (;;) {
+    old_value = method_atomic_fixnum_value(self);
+    new_value = rb_yield(old_value);
+    if (ir_compare_and_set(self, old_value, new_value) == Qtrue) {
+      return new_value;
+    }
+  }
+}

ext/concurrent/atomic_fixnum.h

@@ -9,5 +9,6 @@ VALUE method_atomic_fixnum_value_set(VALUE, VALUE);
 VALUE method_atomic_fixnum_increment(int, VALUE*, VALUE);
 VALUE method_atomic_fixnum_decrement(int, VALUE*, VALUE);
 VALUE method_atomic_fixnum_compare_and_set(VALUE, VALUE, VALUE);
+VALUE method_atomic_fixnum_update(VALUE);
 
 #endif

ext/concurrent/rb_concurrent.c

@@ -52,6 +52,7 @@ void Init_extension() {
   rb_define_method(rb_cAtomicFixnum, "increment", method_atomic_fixnum_increment, -1);
   rb_define_method(rb_cAtomicFixnum, "decrement", method_atomic_fixnum_decrement, -1);
   rb_define_method(rb_cAtomicFixnum, "compare_and_set", method_atomic_fixnum_compare_and_set, 2);
+  rb_define_method(rb_cAtomicFixnum, "update", method_atomic_fixnum_update, 0);
   rb_define_alias(rb_cAtomicFixnum, "up", "increment");
   rb_define_alias(rb_cAtomicFixnum, "down", "decrement");
 }

lib/concurrent/atomic/atomic_fixnum.rb

@@ -110,6 +110,23 @@ def compare_and_set(expect, update)
       end
     end
 
+    # @!macro [attach] atomic_fixnum_method_update
+    #
+    # Pass the current value to the given block, replacing it
+    # with the block's result. May retry if the value changes
+    # during the block's execution.
+    #
+    # @yield [Object] Calculate a new value for the atomic reference using
+    #   given (old) value
+    # @yieldparam [Object] old_value the starting value of the atomic reference
+    #
+    # @return [Object] the new value
+    def update
+      synchronize do
+        @value = yield @value
+      end
+    end
+
     protected
 
     # @!visibility private

lib/concurrent/atomic/read_write_lock.rb

@@ -1,5 +1,5 @@
 require 'thread'
-require 'concurrent/atomic/atomic_reference'
+require 'concurrent/atomic/atomic_fixnum'
 require 'concurrent/errors'
 require 'concurrent/synchronization'
 
@@ -33,7 +33,7 @@ class ReadWriteLock < Synchronization::Object
     WAITING_WRITER = 1 << 15
 
     # @!visibility private
-    RUNNING_WRITER = 1 << 30
+    RUNNING_WRITER = 1 << 29
 
     # @!visibility private
     MAX_READERS    = WAITING_WRITER - 1
@@ -50,11 +50,11 @@ class ReadWriteLock < Synchronization::Object
     # Each reader increments the counter by 1 when acquiring a read lock
     #   (and decrements by 1 when releasing the read lock)
     # The counter is increased by (1 << 15) for each writer waiting to acquire the
-    #   write lock, and by (1 << 30) if the write lock is taken
+    #   write lock, and by (1 << 29) if the write lock is taken
 
     # Create a new `ReadWriteLock` in the unlocked state.
     def initialize
-      @Counter   = AtomicReference.new(0) # single integer which represents lock state
+      @Counter   = AtomicFixnum.new(0) # single integer which represents lock state
       @ReadLock  = Synchronization::Lock.new
       @WriteLock = Synchronization::Lock.new
       ensure_ivar_visibility!
@@ -122,11 +122,11 @@ def acquire_read_lock
             if running_writer?(c)
               @ReadLock.wait_until { !running_writer? }
             else
-              return if @Counter.compare_and_swap(c, c+1)
+              return if @Counter.compare_and_set(c, c+1)
             end
           end
         else
-          break if @Counter.compare_and_swap(c, c+1)
+          break if @Counter.compare_and_set(c, c+1)
         end
       end
       true
@@ -138,7 +138,7 @@ def acquire_read_lock
     def release_read_lock
       while true
         c = @Counter.value
-        if @Counter.compare_and_swap(c, c-1)
+        if @Counter.compare_and_set(c, c-1)
           # If one or more writers were waiting, and we were the last reader, wake a writer up
           if waiting_writer?(c) && running_readers(c) == 1
             @WriteLock.signal
@@ -162,8 +162,8 @@ def acquire_write_lock
 
         if c == 0 # no readers OR writers running
           # if we successfully swap the RUNNING_WRITER bit on, then we can go ahead
-          break if @Counter.compare_and_swap(0, RUNNING_WRITER)
-        elsif @Counter.compare_and_swap(c, c+WAITING_WRITER)
+          break if @Counter.compare_and_set(0, RUNNING_WRITER)
+        elsif @Counter.compare_and_set(c, c+WAITING_WRITER)
           while true
             # Now we have successfully incremented, so no more readers will be able to increment
             #   (they will wait instead)
@@ -180,7 +180,7 @@ def acquire_write_lock
             # Then we are OK to stop waiting and go ahead
             # Otherwise go back and wait again
             c = @Counter.value
-            break if !running_writer?(c) && !running_readers?(c) && @Counter.compare_and_swap(c, c+RUNNING_WRITER-WAITING_WRITER)
+            break if !running_writer?(c) && !running_readers?(c) && @Counter.compare_and_set(c, c+RUNNING_WRITER-WAITING_WRITER)
           end
           break
         end