TVar: initial global lock implementation.

Author: chrisseaton

lib/concurrent.rb

@@ -22,6 +22,7 @@
 require 'concurrent/stoppable'
 require 'concurrent/supervisor'
 require 'concurrent/timer_task'
+require 'concurrent/tvar'
 require 'concurrent/utilities'
 
 require 'concurrent/global_thread_pool'

lib/concurrent/tvar.rb

@@ -0,0 +1,148 @@
+module Concurrent
+
+  ABORTED = Object.new
+
+  class TVar
+
+    def initialize(value)
+      @value = value
+    end
+
+    def value
+      Concurrent::atomically do
+        Transaction::current.read(self)
+      end
+    end
+
+    def value=(value)
+      Concurrent::atomically do
+        Transaction::current.write(self, value)
+      end
+    end
+
+    def unsafe_value
+      @value
+    end
+
+    def unsafe_value=(value)
+      @value = value
+    end
+
+  end
+
+  UndoLogEntry = Struct.new(:tvar, :value)
+
+  class Transaction
+
+    LOCK = Mutex.new
+
+    def initialize
+      @undo_log = []
+
+      LOCK.lock
+    end
+
+    def read(tvar)
+      validate
+      tvar.unsafe_value
+    end
+
+    def write(tvar, value)
+      @undo_log.push(UndoLogEntry.new(tvar, tvar.unsafe_value))
+      tvar.unsafe_value = value
+    end
+
+    def abort
+      @undo_log.each do |entry|
+        entry.tvar.unsafe_value = entry.value
+      end
+
+      unlock
+    end
+
+    def commit
+      validate
+      unlock
+      true
+    end
+
+    def validate
+    end
+
+    def unlock
+      LOCK.unlock
+    end
+
+    def self.current
+      Thread.current.thread_variable_get(:transaction)
+    end
+
+    def self.current=(transaction)
+      Thread.current.thread_variable_set(:transaction, transaction)
+    end
+
+  end
+
+  class AbortException < Exception
+  end
+
+  def atomically
+    raise ArgumentError.new('no block given') unless block_given?
+
+    # Get the current transaction
+
+    transaction = Transaction::current
+
+    # Are we not already in a transaction (not nested)?
+
+    if transaction.nil?
+      # New transaction
+
+      begin
+        # Retry loop
+        
+        loop do
+
+          # Create a new transaction
+
+          transaction = Transaction.new
+          Transaction::current = transaction
+
+          # Run the block, aborting on exceptions
+
+          begin
+            result = yield
+          rescue AbortException => e
+            transaction.abort
+            result = ABORTED
+          rescue => e
+            transaction.abort
+            throw e
+          end
+          # If we can commit, break out of the loop
+
+          if result != ABORTED
+            if transaction.commit
+              break result
+            end
+          end
+        end
+      ensure
+        # Clear the current transaction
+
+        Transaction::current = nil
+      end
+    else
+      # Nested transaction - flatten it and just run the block
+
+      yield
+    end
+  end
+
+  def abort_transaction
+    raise AbortException.new
+  end
+
+  module_function :atomically, :abort_transaction
+
+end

md/tvar.md

@@ -0,0 +1,31 @@
+# TVars
+
+`TVar` and `atomically` implement a transactional memory. A `TVar` is a single
+item container that always contains exactly one value. The `atomically` method
+allows you to modify a set of `TVar` objects with the guarantee that all of the
+updates are collectively atomic - they either all happen or none of them do -
+consistent - a `TVar` will never enter an illegal state - and isolated - atomic
+blocks never interfere with each other when they are running. You may recognise
+these properties from database transactions.
+
+There are some very important and unusual semantics that you must be aware of:
+
+*   Most importantly, the block that you pass to `atomically` may be executed more
+than once. In most cases your code should be free of side-effects, except for
+via `TVar`.
+
+*   If an exception escapes an `atomically` block it will abort the transaction.
+
+*   It is undefined behaviour to use `callcc` or `Fiber` with `atomically`.
+
+*   If you create a new thread within an `atomically`, it will not be part of
+the transaction. Creating a thread counts as a side-effect.
+
+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.
+
+See:
+
+1.  T. Harris, J. Larus, and R. Rajwar. Transactional Memory. Morgan & Claypool, second edition, 2010.

spec/concurrent/tvar_spec.rb

@@ -0,0 +1,137 @@
+require 'spec_helper'
+
+module Concurrent
+
+  describe TVar do
+
+    context '#initialize' do
+
+      it 'accepts an initial value' do
+        t = TVar.new(14)
+        t.value.should eq 14
+      end
+
+    end
+
+    context '#value' do
+
+      it 'gets the value' do
+        t = TVar.new(14)
+        t.value.should eq 14
+      end
+
+    end
+
+    context '#value=' do
+
+      it 'sets the value' do
+        t = TVar.new(14)
+        t.value = 2
+        t.value.should eq 2
+      end
+
+    end
+
+  end
+
+  describe '#atomically' do
+
+    it 'raises an exception when no block given' do
+      expect { Concurrent::atomically }.to raise_error(ArgumentError)
+    end
+
+    it 'retries on abort' do
+      count = 0
+
+      Concurrent::atomically do
+        if count == 0
+          count = 1
+          Concurrent::abort_transaction
+        else
+          count = 2
+        end
+      end
+
+      count.should eq 2
+    end
+
+    it 'commits writes if the transaction succeeds' do
+      t = TVar.new(0)
+
+      Concurrent::atomically do
+        t.value = 1
+      end
+
+      t.value.should eq 1
+    end
+
+    it 'undoes writes if the transaction is aborted' do
+      t = TVar.new(0)
+
+      count = 0
+
+      Concurrent::atomically do
+        if count == 0
+          t.value = 1
+          count = 1
+          Concurrent::abort_transaction
+        end
+      end
+
+      t.value.should eq 0
+    end
+
+    it 'provides atomicity' do
+      t1 = TVar.new(0)
+      t2 = TVar.new(0)
+
+      count = 0
+
+      Concurrent::atomically do
+        if count == 0
+          count = 1
+          t1.value = 1
+          Concurrent::abort_transaction
+          t2.value = 2
+        end
+      end
+
+      t1.value.should eq 0
+      t2.value.should eq 0
+    end
+
+    it 'provides isolation' do
+      t = TVar.new(0)
+
+      Thread.new do
+        Concurrent::atomically do
+          t1.value = 1
+          sleep(1)
+        end
+      end
+
+      sleep(0.5)
+
+      t.value.should eq 0
+    end
+
+    it 'nests' do
+      Concurrent::atomically do
+        Concurrent::atomically do
+          Concurrent::atomically do
+          end
+        end
+      end
+    end
+
+  end
+
+  describe '#abort_transaction' do
+
+    it 'raises an exception outside an #atomically block' do
+      expect { Concurrent::abort_transaction }.to raise_error(Concurrent::AbortException)
+    end
+
+  end
+
+end