Simplify TVar implementation at the cost of performance

Author: chrisseaton

docs-source/tvar.md

@@ -24,16 +24,9 @@ We implement nested transactions by flattening.
 We only support strong isolation if you use the API correctly. In order words,
 we do not support strong isolation.
 
-Our implementation uses a very simple two-phased locking with versioned locks
-algorithm and lazy writes, as per [1].
-
-See:
-
-1.  T. Harris, J. Larus, and R. Rajwar. Transactional Memory. Morgan & Claypool, second edition, 2010.
-
-Note that this implementation allows transactions to continue in a zombie state
-with inconsistent reads, so it's possible for the marked exception to be raised
-in the example below.
+Our implementation uses a very simple algorithm that locks each `TVar` when it
+is first read or written. If it cannot lock a `TVar` it aborts and retries.
+There is no contention manager so competing transactions may retry eternally.
 
 ```ruby
 require 'concurrent-ruby'

lib/concurrent-ruby/concurrent/tvar.rb

@@ -15,7 +15,6 @@ class TVar < Synchronization::Object
     # Create a new `TVar` with an initial value.
     def initialize(value)
       @value = value
-      @version = 0
       @lock = Mutex.new
     end
 
@@ -43,16 +42,6 @@ def unsafe_value=(value) # :nodoc:
       @value = value
     end
 
-    # @!visibility private
-    def unsafe_version # :nodoc:
-      @version
-    end
-
-    # @!visibility private
-    def unsafe_increment_version # :nodoc:
-      @version += 1
-    end
-
     # @!visibility private
     def unsafe_lock # :nodoc:
       @lock
@@ -164,86 +153,57 @@ class Transaction
 
     ABORTED = ::Object.new
 
-    ReadLogEntry = Struct.new(:tvar, :version)
+    OpenEntry = Struct.new(:value, :modified)
 
     AbortError = Class.new(StandardError)
     LeaveError = Class.new(StandardError)
 
     def initialize
-      @read_log  = []
-      @write_log = {}
+      @open_tvars = {}
     end
 
     def read(tvar)
-      Concurrent::abort_transaction unless valid?
-
-      if @write_log.has_key? tvar
-        @write_log[tvar]
-      else
-        @read_log.push(ReadLogEntry.new(tvar, tvar.unsafe_version))
-        tvar.unsafe_value
-      end
+      entry = open(tvar)
+      entry.value
     end
 
     def write(tvar, value)
-      # Have we already written to this TVar?
+      entry = open(tvar)
+      entry.modified = true
+      entry.value = value
+    end
 
-      if @write_log.has_key? tvar
-        # Record the value written
-        @write_log[tvar] = value
-      else
-        # Try to lock the TVar
+    def open(tvar)
+      entry = @open_tvars[tvar]
 
+      unless entry
         unless tvar.unsafe_lock.try_lock
-          # Someone else is writing to this TVar - abort
           Concurrent::abort_transaction
         end
 
-        # Record the value written
-  
-        @write_log[tvar] = value
-
-        # If we previously read from it, check the version hasn't changed
-
-        @read_log.each do |log_entry|
-          if log_entry.tvar == tvar and tvar.unsafe_version > log_entry.version
-            Concurrent::abort_transaction
-          end
-        end
+        entry = OpenEntry.new(tvar.unsafe_value, false)
+        @open_tvars[tvar] = entry
       end
+
+      entry
     end
 
     def abort
       unlock
     end
 
     def commit
-      return false unless valid?
-
-      @write_log.each_pair do |tvar, value|
-        tvar.unsafe_value = value
-        tvar.unsafe_increment_version
-      end
-
-      unlock
-
-      true
-    end
-
-    def valid?
-      @read_log.each do |log_entry|
-        unless @write_log.has_key? log_entry.tvar
-          if log_entry.tvar.unsafe_version > log_entry.version
-            return false
-          end
+      @open_tvars.each do |tvar, entry|
+        if entry.modified
+          tvar.unsafe_value = entry.value
         end
       end
 
-      true
+      unlock
     end
 
     def unlock
-      @write_log.each_key do |tvar|
+      @open_tvars.each_key do |tvar|
         tvar.unsafe_lock.unlock
       end
     end

spec/concurrent/tvar_spec.rb

@@ -106,48 +106,6 @@ module Concurrent
       expect(t2.value).to eq 0
     end
 
-    it 'provides weak isolation' do
-      t = TVar.new(0)
-
-      a = CountDownLatch.new
-      b = CountDownLatch.new
-
-      in_thread do
-        Concurrent::atomically do
-          t.value = 1
-          a.count_down
-          b.wait
-          Concurrent.leave_transaction
-        end
-      end
-
-      Concurrent::atomically do
-        a.wait
-        expect(t.value).to eq 0
-        b.count_down
-        Concurrent.leave_transaction
-      end
-    end
-
-    it 'is implemented with lazy writes' do
-      t = TVar.new(0)
-
-      a = CountDownLatch.new
-      b = CountDownLatch.new
-
-      in_thread do
-        Concurrent::atomically do
-          t.value = 1
-          a.count_down
-          b.wait
-        end
-      end
-
-      a.wait
-      expect(t.value).to eq 0
-      b.count_down
-    end
-
     it 'nests' do
       t = TVar.new(0)
 

test/stress/tvar/atomicity.rb

@@ -0,0 +1,31 @@
+require 'concurrent-ruby'
+
+v1 = Concurrent::TVar.new(0)
+v2 = Concurrent::TVar.new(0)
+
+Thread.new do
+  loop do
+    Concurrent.atomically do
+      v1.value += 1
+      v2.value += 1
+    end
+  end
+end
+
+loop do
+  a, b = Concurrent.atomically {
+    a = v1.value
+    b = v2.value
+    [a, b]
+  }
+  raise if a != b
+  p a
+
+  a, b = Concurrent.atomically {
+    b = v2.value
+    a = v1.value
+    [a, b]
+  }
+  raise if a != b
+  p a
+end

test/stress/tvar/opacity.rb

@@ -0,0 +1,31 @@
+require 'concurrent-ruby'
+
+v1 = Concurrent::TVar.new(0)
+v2 = Concurrent::TVar.new(0)
+
+Thread.new do
+  loop do
+    Concurrent.atomically do
+      v1.value += 1
+      v2.value += 1
+    end
+  end
+end
+
+loop do
+  a, b = Concurrent.atomically {
+    a = v1.value
+    b = v2.value
+    raise if a != b
+    [a, b]
+  }
+  p a
+
+  a, b = Concurrent.atomically {
+    b = v2.value
+    a = v1.value
+    raise if a != b
+    [a, b]
+  }
+  p a
+end