Make CRuby Set thread safe

lib/concurrent-ruby/concurrent/set.rb

@@ -23,9 +23,21 @@ module Concurrent
   # @!macro internal_implementation_note
   SetImplementation = case
                       when Concurrent.on_cruby?
-                        # Because MRI never runs code in parallel, the existing
-                        # non-thread-safe structures should usually work fine.
-                        ::Set
+                        # The CRuby implementation of Set is written in Ruby itself and is
+                        # not thread safe for certain methods.
+                        require 'monitor'
+                        require 'concurrent/thread_safe/util/data_structures'
+
+                        class CRubySet < ::Set
+                          def initialize(*args)
+                            self.send(:_mon_initialize)
+                            super(*args)
+                          end
+                        end
+
+                        # make use of the same RBX
+                        ThreadSafe::Util.make_synchronized_on_rbx Concurrent::CRubySet
+                        CRubySet
 
                       when Concurrent.on_jruby?
                         require 'jruby/synchronized'

lib/concurrent-ruby/concurrent/thread_safe/util/data_structures.rb

@@ -17,7 +17,7 @@ def self.make_synchronized_on_rbx(klass)
           private
 
           def _mon_initialize
-            @_monitor = Monitor.new unless @_monitor # avoid double initialisation
+            @_monitor ||= Monitor.new # avoid double initialisation
           end
 
           def self.new(*args)

spec/concurrent/set_spec.rb

@@ -42,7 +42,7 @@ module Concurrent
     end
 
     context 'concurrency' do
-      it do
+      it 'simple' do
         (1..Concurrent::ThreadSafe::Test::THREADS).map do |i|
           in_thread do
             1000.times do
@@ -55,6 +55,33 @@ module Concurrent
         end.map(&:join)
         expect(set).to be_empty
       end
+
+      it 'force context switch' do
+        barrier = Concurrent::CyclicBarrier.new(2)
+
+        # methods like include? or delete? are implemented for CRuby in Ruby itself
+        # @see https://github.com/ruby/ruby/blob/master/lib/set.rb
+        set.clear
+
+        # add a single element
+        set.add(1)
+
+        # This thread should start and `Set#reject!` in CRuby should cache a value of `0` for size
+        thread_reject = in_thread do
+          # we expect this to return nil since nothing should have changed.
+          expect(set.reject! do |v|
+            barrier.wait
+            v == 1 # only delete the 1 value
+          end).to eq set
+        end
+
+        thread_add = in_thread do
+          barrier.wait
+          expect(set.add?(1)).to eq set
+        end
+
+        join_with [thread_reject, thread_add]
+      end
     end
   end
 end