diff --git a/lib/concurrent-ruby/concurrent/set.rb b/lib/concurrent-ruby/concurrent/set.rb index 602d49408..0cb26e769 100644 --- a/lib/concurrent-ruby/concurrent/set.rb +++ b/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' diff --git a/lib/concurrent-ruby/concurrent/thread_safe/util/data_structures.rb b/lib/concurrent-ruby/concurrent/thread_safe/util/data_structures.rb index ff1e8ed97..45a20cadc 100644 --- a/lib/concurrent-ruby/concurrent/thread_safe/util/data_structures.rb +++ b/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) diff --git a/spec/concurrent/set_spec.rb b/spec/concurrent/set_spec.rb index 7c69b4bec..d9e657fc9 100644 --- a/spec/concurrent/set_spec.rb +++ b/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