Add java implementation of JavaSynchronizedObject

Author: pitr-ch

ext/ConcurrentRubyExtService.java

@@ -9,6 +9,7 @@ public boolean basicLoad(final Ruby runtime) throws IOException {
         new com.concurrent_ruby.ext.JavaAtomicBooleanLibrary().load(runtime, false);
         new com.concurrent_ruby.ext.JavaAtomicFixnumLibrary().load(runtime, false);
         new com.concurrent_ruby.ext.JavaSemaphoreLibrary().load(runtime, false);
+        new com.concurrent_ruby.ext.JavaSynchronizedObjectLibrary().load(runtime, false);
         return true;
     }
 }

ext/com/concurrent_ruby/ext/JavaSynchronizedObjectLibrary.java

@@ -0,0 +1,100 @@
+package com.concurrent_ruby.ext;
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.jruby.Ruby;
+import org.jruby.RubyClass;
+import org.jruby.RubyModule;
+import org.jruby.RubyObject;
+import org.jruby.anno.JRubyClass;
+import org.jruby.anno.JRubyMethod;
+import org.jruby.runtime.ObjectAllocator;
+import org.jruby.runtime.builtin.IRubyObject;
+import org.jruby.runtime.load.Library;
+import org.jruby.runtime.Block;
+import org.jruby.RubyBoolean;
+import org.jruby.RubyNil;
+import org.jruby.runtime.ThreadContext;
+
+public class JavaSynchronizedObjectLibrary implements Library {
+
+    public void load(Ruby runtime, boolean wrap) throws IOException {
+        RubyModule concurrentMod = runtime.defineModule("Concurrent");
+        RubyClass javaSynchronizedObjectClass = concurrentMod.defineClassUnder(
+                "JavaSynchronizedObject",
+                concurrentMod.getClass("AbstractSynchronizedObject"),
+                JRUBYREFERENCE_ALLOCATOR);
+        javaSynchronizedObjectClass.defineAnnotatedMethods(JavaSynchronizedObject.class);
+    }
+
+    private static final ObjectAllocator JRUBYREFERENCE_ALLOCATOR = new ObjectAllocator() {
+        public IRubyObject allocate(Ruby runtime, RubyClass klazz) {
+            return new JavaSynchronizedObject(runtime, klazz);
+        }
+    };
+
+    @JRubyClass(name = "JavaSynchronizedObject", parent = "AbstractSynchronizedObject")
+    public static class JavaSynchronizedObject extends RubyObject {
+
+        public JavaSynchronizedObject(Ruby runtime, RubyClass metaClass) {
+            super(runtime, metaClass);
+        }
+
+        @JRubyMethod
+        public IRubyObject initialize(ThreadContext context) {
+            return context.nil;
+        }
+
+        @JRubyMethod(name = "synchronize")
+        public IRubyObject rubySynchronize(ThreadContext context, Block block) {
+            synchronized (this) {
+                return block.yield(context, null);
+            }
+        }
+
+        @JRubyMethod(name = "ns_wait", optional = 1)
+        public IRubyObject nsWait(ThreadContext context, IRubyObject[] args) {
+            Ruby runtime = context.runtime;
+            if (args.length > 1) {
+                throw runtime.newArgumentError(args.length, 1);
+            }
+            Double timeout = null;
+            if (args.length > 0 && !args[0].isNil()) {
+                timeout = args[0].convertToFloat().getDoubleValue();
+                if (timeout < 0) {
+                    throw runtime.newArgumentError("time interval must be positive");
+                }
+            }
+            if (Thread.interrupted()) {
+                throw runtime.newConcurrencyError("thread interrupted");
+            }
+            boolean success = false;
+            try {
+                success = context.getThread().wait_timeout(this, timeout);
+            } catch (InterruptedException ie) {
+                throw runtime.newConcurrencyError(ie.getLocalizedMessage());
+            } finally {
+                // An interrupt or timeout may have caused us to miss
+                // a notify that we consumed, so do another notify in
+                // case someone else is available to pick it up.
+                if (!success) {
+                    this.notify();
+                }
+            }
+            return this;
+        }
+
+        @JRubyMethod(name = "ns_signal")
+        public IRubyObject nsSignal(ThreadContext context) {
+            notify();
+            return this;
+        }
+
+        @JRubyMethod(name = "ns_broadcast")
+        public IRubyObject nsBroadcast(ThreadContext context) {
+            notifyAll();
+            return this;
+        }
+    }
+}

lib/concurrent.rb

@@ -1,5 +1,7 @@
 require 'concurrent/version'
 
+require 'concurrent/synchronized_object'
+
 require 'concurrent/configuration'
 
 require 'concurrent/actor'
@@ -9,7 +11,6 @@
 require 'concurrent/errors'
 require 'concurrent/executors'
 require 'concurrent/utilities'
-require 'concurrent/synchronized_object'
 
 require 'concurrent/atomic'
 require 'concurrent/agent'

lib/concurrent/atomic/cyclic_barrier.rb

@@ -20,10 +20,9 @@ def initialize(parties, &block)
         raise ArgumentError.new('count must be in integer greater than or equal zero')
       end
       synchronize do
-        @parties        = parties
-        @number_waiting = 0
-        @action         = block
-        @generation     = Generation.new(:waiting)
+        @parties = parties
+        @action  = block
+        ns_next_generation
       end
     end
 
@@ -54,10 +53,16 @@ def wait(timeout = nil)
 
         if @number_waiting == @parties
           @action.call if @action
-          set_status_and_restore(:fulfilled)
+          ns_generation_done @generation, :fulfilled
           true
         else
-          wait_for_wake_up(@generation, timeout)
+          generation = @generation
+          if ns_wait_until(timeout) { generation.status != :waiting }
+            generation.status == :fulfilled
+          else
+            ns_generation_done generation, :broken, false
+            false
+          end
         end
       end
     end
@@ -70,7 +75,7 @@ def wait(timeout = nil)
     #
     # @return [nil]
     def reset
-      synchronize { set_status_and_restore(:reset) }
+      synchronize { ns_generation_done @generation, :reset }
     end
 
     # A barrier can be broken when:
@@ -85,21 +90,17 @@ def broken?
 
     private
 
-    def set_status_and_restore(new_status)
-      @generation.status = new_status
+    def ns_generation_done(generation, status, continue = true)
+      generation.status = status
+      ns_next_generation if continue
       ns_broadcast
+    end
+
+    def ns_next_generation
       @generation     = Generation.new(:waiting)
       @number_waiting = 0
     end
 
-    def wait_for_wake_up(generation, timeout)
-      if ns_wait_until(timeout) { generation.status != :waiting }
-        generation.status == :fulfilled
-      else
-        generation.status = :broken
-        ns_broadcast
-        false
-      end
-    end
+
   end
 end

lib/concurrent/extension_helper.rb

@@ -2,6 +2,9 @@
 
 module Concurrent
 
+  class AbstractSynchronizedObject # FIXME has to be present before Java extensions are loaded
+  end
+
   @@c_ext_loaded ||= false
   @@java_ext_loaded ||= false
 

lib/concurrent/synchronized_object.rb

@@ -1,3 +1,5 @@
+require 'concurrent/utility/engine'
+
 module Concurrent
 
   # Safe synchronization under any Ruby implementation.
@@ -75,7 +77,7 @@ def broadcast
     def ns_wait_until(timeout, &condition)
       if timeout
         wait_until = Concurrent.monotonic_time + timeout
-        while true
+        loop do
           now              = Concurrent.monotonic_time
           condition_result = condition.call
           # 0.001 correction to avoid error when `wait_until - now` is smaller than 0.0005 and rounded to 0
@@ -109,8 +111,7 @@ def ns_broadcast
   if Concurrent.on_jruby?
     require 'jruby'
 
-    # roughly more than 2x faster
-    class JavaSynchronizedObject < AbstractSynchronizedObject
+    class JavaPureSynchronizedObject < AbstractSynchronizedObject
       def initialize
       end
 
@@ -121,8 +122,10 @@ def synchronize
       private
 
       def ns_wait(timeout = nil)
-        JRuby.reference0(Thread.current).wait_timeout(self, timeout)
+        success = JRuby.reference0(Thread.current).wait_timeout(JRuby.reference0(self), timeout)
         self
+      ensure
+        ns_signal unless success
       end
 
       def ns_broadcast

spec/concurrent/atomic/cyclic_barrier_spec.rb

@@ -173,6 +173,15 @@ module Concurrent
             expect(barrier).to be_broken
           end
 
+          it 'breaks the barrier and release all other threads 2' do
+            t1 = Thread.new { barrier.wait(0.1) }
+            t2 = Thread.new { barrier.wait(0.1) }
+
+            [t1, t2].each(&:join)
+
+            expect(barrier).to be_broken
+          end
+
           it 'does not execute the block on timeout' do
             counter = AtomicFixnum.new
             barrier = described_class.new(parties) { counter.increment }
@@ -242,5 +251,4 @@ def barrier.simulate_spurious_wake_up
       end
     end
   end
-
 end