Merge pull request #357 from ruby-concurrency/synchronization

Synchronisation fixes

Author: jdantonio

README.md

@@ -39,6 +39,7 @@
 
 MRI 1.9.3, 2.0, 2.1, 2.2, JRuby (1.9 mode), and Rubinius 2.x are supported.
 This gem should be fully compatible with any interpreter that is compliant with Ruby 1.9.3 or newer.
+Java 8 is required for JRuby (Java 7 support is deprecated in version 0.9 and will be removed in 1.0).
 
 ## Features & Documentation
 

doc/synchronization.md

@@ -139,13 +139,14 @@ We are interested in following behaviors:
 
 ### Variables
 
--   **Local variables** - atomic assignment, non-volatile. 
+-   **Local variables** - atomic assignment (only Integer and Object), non-volatile. 
     -   Consequence: a lambda defined on `thread1` executing on `thread2` may not see updated values in local variables captured in its closure.
     -   Reason: local variables are non-volatile on Jruby and Rubinius.
--   **Instance variables** - atomic assignment, non-volatile. 
+-   **Instance variables** - atomic assignment (only Integer and Object), non-volatile. 
     -   Consequence: Different thread may see old values; different thread may see not fully-initialized object.
     -   Reason: local variables are non-volatile on Jruby and Rubinius.
 -   **Constants** - atomic assignment, volatile.
+-   **Assignments of Float** may not be atomic (some implementations (e.g. Truffle) may use native double).
 
 Other:
 

ext/com/concurrent_ruby/ext/SynchronizationLibrary.java

@@ -21,6 +21,12 @@
 
 public class SynchronizationLibrary implements Library {
 
+    private static final ObjectAllocator JRUBYREFERENCE_ALLOCATOR = new ObjectAllocator() {
+        public IRubyObject allocate(Ruby runtime, RubyClass klazz) {
+            return new JavaObject(runtime, klazz);
+        }
+    };
+
     public void load(Ruby runtime, boolean wrap) throws IOException {
         RubyModule synchronizationModule = runtime.
                 defineModule("Concurrent").
@@ -36,15 +42,13 @@ public void load(Ruby runtime, boolean wrap) throws IOException {
         synchronizedObjectJavaClass.defineAnnotatedMethods(JavaObject.class);
     }
 
-    private static final ObjectAllocator JRUBYREFERENCE_ALLOCATOR = new ObjectAllocator() {
-        public IRubyObject allocate(Ruby runtime, RubyClass klazz) {
-            return new JavaObject(runtime, klazz);
-        }
-    };
-
     @JRubyClass(name = "JavaObject", parent = "AbstractObject")
     public static class JavaObject extends RubyObject {
 
+        public static final long AN_VOLATILE_FIELD_OFFSET =
+                UnsafeHolder.fieldOffset(JavaObject.class, "anVolatileField");
+        private volatile int anVolatileField = 0;
+
         public JavaObject(Ruby runtime, RubyClass metaClass) {
             super(runtime, metaClass);
         }
@@ -107,24 +111,30 @@ public IRubyObject nsBroadcast(ThreadContext context) {
 
         @JRubyMethod(name = "ensure_ivar_visibility!", visibility = Visibility.PROTECTED)
         public IRubyObject ensureIvarVisibilityBang(ThreadContext context) {
-            if (UnsafeHolder.SUPPORTS_FENCES)
-                UnsafeHolder.storeFence();
-            else
-                anVolatileField = 1;
+            if (UnsafeHolder.U == null) {
+                // We are screwed
+                throw new UnsupportedOperationException();
+            } else if (UnsafeHolder.SUPPORTS_FENCES)
+                // We have to prevent ivar writes to reordered with storing of the final instance reference
+                // Therefore wee need a fullFence to prevent reordering in both directions.
+                UnsafeHolder.fullFence();
+            else {
+                // Assumption that this is not eliminated, if false it will break non x86 platforms.
+                UnsafeHolder.U.putIntVolatile(this, AN_VOLATILE_FIELD_OFFSET, 1);
+                UnsafeHolder.U.getIntVolatile(this, AN_VOLATILE_FIELD_OFFSET);
+            }
             return context.nil;
         }
 
-        private volatile int anVolatileField = 0; // TODO unused on JAVA8
-        public static final long AN_VOLATILE_FIELD_OFFSET =
-                UnsafeHolder.fieldOffset(JavaObject.class, "anVolatileField");
-
         @JRubyMethod(name = "instance_variable_get_volatile", visibility = Visibility.PROTECTED)
         public IRubyObject instanceVariableGetVolatile(ThreadContext context, IRubyObject name) {
             if (UnsafeHolder.U == null) {
+                // TODO: Possibly dangerous, there may be a deadlock on the this
                 synchronized (this) {
                     return instance_variable_get(context, name);
                 }
             } else if (UnsafeHolder.SUPPORTS_FENCES) {
+                // ensure we see latest value
                 UnsafeHolder.loadFence();
                 return instance_variable_get(context, name);
             } else {
@@ -136,16 +146,18 @@ public IRubyObject instanceVariableGetVolatile(ThreadContext context, IRubyObjec
         @JRubyMethod(name = "instance_variable_set_volatile", visibility = Visibility.PROTECTED)
         public IRubyObject InstanceVariableSetVolatile(ThreadContext context, IRubyObject name, IRubyObject value) {
             if (UnsafeHolder.U == null) {
+                // TODO: Possibly dangerous, there may be a deadlock on the this
                 synchronized (this) {
                     return instance_variable_set(name, value);
                 }
             } else if (UnsafeHolder.SUPPORTS_FENCES) {
-                IRubyObject result = instance_variable_set(name, value);
+                final IRubyObject result = instance_variable_set(name, value);
+                // ensure we make latest value visible
                 UnsafeHolder.storeFence();
                 return result;
             } else {
+                final IRubyObject result = instance_variable_set(name, value);
                 UnsafeHolder.U.putIntVolatile(this, AN_VOLATILE_FIELD_OFFSET, 1);
-                IRubyObject result = instance_variable_set(name, value);
                 return result;
             }
         }

lib/concurrent/concern/deprecation.rb

@@ -10,18 +10,25 @@ module Deprecation
       include Concern::Logging
 
       def deprecated(message, strip = 2)
-        caller_line = caller(strip).first
-        klass       = if Class === self
+        caller_line = caller(strip).first if strip > 0
+        klass       = if Module === self
                         self
                       else
                         self.class
                       end
-        log WARN, klass.to_s, format("[DEPRECATED] %s\ncalled on: %s", message, caller_line)
+        message     = if strip > 0
+                        format("[DEPRECATED] %s\ncalled on: %s", message, caller_line)
+                      else
+                        format('[DEPRECATED] %s', message)
+                      end
+        log WARN, klass.to_s, message
       end
 
       def deprecated_method(old_name, new_name)
         deprecated "`#{old_name}` is deprecated and it'll removed in next release, use `#{new_name}` instead", 3
       end
+
+      extend self
     end
   end
 end

lib/concurrent/concern/logging.rb

@@ -1,4 +1,5 @@
 require 'logger'
+require 'concurrent/configuration'
 
 module Concurrent
   module Concern

lib/concurrent/configuration.rb

@@ -11,37 +11,6 @@ module Concurrent
   extend Concern::Logging
   extend Concern::Deprecation
 
-  # Suppresses all output when used for logging.
-  NULL_LOGGER = lambda { |level, progname, message = nil, &block| }
-
-  # @!visibility private
-  GLOBAL_LOGGER = AtomicReference.new(NULL_LOGGER)
-  private_constant :GLOBAL_LOGGER
-
-  # @!visibility private
-  GLOBAL_FAST_EXECUTOR = Delay.new { Concurrent.new_fast_executor(auto_terminate: true) }
-  private_constant :GLOBAL_FAST_EXECUTOR
-
-  # @!visibility private
-  GLOBAL_IO_EXECUTOR = Delay.new { Concurrent.new_io_executor(auto_terminate: true) }
-  private_constant :GLOBAL_IO_EXECUTOR
-
-  # @!visibility private
-  GLOBAL_TIMER_SET = Delay.new { TimerSet.new(auto_terminate: true) }
-  private_constant :GLOBAL_TIMER_SET
-
-  # @!visibility private
-  GLOBAL_IMMEDIATE_EXECUTOR = ImmediateExecutor.new
-  private_constant :GLOBAL_IMMEDIATE_EXECUTOR
-
-  def self.global_logger
-    GLOBAL_LOGGER.value
-  end
-
-  def self.global_logger=(value)
-    GLOBAL_LOGGER.value = value
-  end
-
   # @return [Logger] Logger with provided level and output.
   def self.create_stdlib_logger(level = Logger::FATAL, output = $stderr)
     logger           = Logger.new(output)
@@ -62,17 +31,48 @@ def self.create_stdlib_logger(level = Logger::FATAL, output = $stderr)
              progname,
              formatted_message
     end
-    logger
+
+    lambda do |level, progname, message = nil, &block|
+      logger.add level, message, progname, &block
+    end
   end
 
   # Use logger created by #create_stdlib_logger to log concurrent-ruby messages.
   def self.use_stdlib_logger(level = Logger::FATAL, output = $stderr)
-    logger = create_stdlib_logger level, output
-    Concurrent.global_logger = lambda do |level, progname, message = nil, &block|
-      logger.add level, message, progname, &block
-    end
+    Concurrent.global_logger = create_stdlib_logger level, output
   end
 
+  # Suppresses all output when used for logging.
+  NULL_LOGGER   = lambda { |level, progname, message = nil, &block| }
+
+  # @!visibility private
+  GLOBAL_LOGGER = AtomicReference.new(create_stdlib_logger(Logger::WARN))
+  private_constant :GLOBAL_LOGGER
+
+  def self.global_logger
+    GLOBAL_LOGGER.value
+  end
+
+  def self.global_logger=(value)
+    GLOBAL_LOGGER.value = value
+  end
+
+  # @!visibility private
+  GLOBAL_FAST_EXECUTOR = Delay.new { Concurrent.new_fast_executor(auto_terminate: true) }
+  private_constant :GLOBAL_FAST_EXECUTOR
+
+  # @!visibility private
+  GLOBAL_IO_EXECUTOR = Delay.new { Concurrent.new_io_executor(auto_terminate: true) }
+  private_constant :GLOBAL_IO_EXECUTOR
+
+  # @!visibility private
+  GLOBAL_TIMER_SET = Delay.new { TimerSet.new(auto_terminate: true) }
+  private_constant :GLOBAL_TIMER_SET
+
+  # @!visibility private
+  GLOBAL_IMMEDIATE_EXECUTOR = ImmediateExecutor.new
+  private_constant :GLOBAL_IMMEDIATE_EXECUTOR
+
   # Disables AtExit handlers including pool auto-termination handlers.
   # When disabled it will be the application programmer's responsibility
   # to ensure that the handlers are shutdown properly prior to application
@@ -275,4 +275,18 @@ def self.configuration
   def self.configure
     yield(configuration)
   end
+
+  # for dependency reasons this check cannot be in concurrent/synchronization
+  if Concurrent.on_jruby?
+    require 'java'
+
+    version_string = java.lang.System.getProperties['java.runtime.version']
+    version        = version_string.split('.', 3)[0..1].map(&:to_i)
+    if (version <=> [1, 8]) < 0
+      deprecated <<-TXT.gsub(/^\s*\|/, '').chop, 0
+          |Java 7 is deprecated, please use Java 8.
+          |Java 7 support is only best effort, it may not work. It will be removed in next release (1.0).
+      TXT
+    end
+  end
 end