wip: Initial import of ruby-atomic.

Author: jdantonio

.gitignore

@@ -1,5 +1,4 @@
 Gemfile.lock
-tests.txt
 *.gem
 .rvmrc
 .ruby-version
@@ -21,9 +20,13 @@ coverage
 .DS_Store
 TAGS
 tmtags
-*.swo
-*.swp
+*.sw?
 .idea
 .rbx/*
-*.py
-*.pyc
+lib/*.jar
+lib/*.bundle
+ext/*.bundle
+ext/*.so
+ext/*.jar
+pkg
+*.gem

Gemfile

@@ -2,13 +2,13 @@ source 'https://rubygems.org'
 
 gemspec
 
-
 group :development do
   gem 'rake', '~> 10.2.2'
   gem 'countloc', '~> 0.4.0', platforms: :mri
   gem 'yard', '~> 0.8.7.4'
   gem 'inch', '~> 0.4.1', platforms: :mri
   gem 'redcarpet', platforms: :mri # understands github markdown
+  gem 'rake-compiler', '~> 0.9.2', platforms: [:mri, :mswin, :mingw]
 end
 
 group :testing do

Rakefile

@@ -1,12 +1,9 @@
-$:.push File.join(File.dirname(__FILE__), 'lib')
-$:.push File.join(File.dirname(__FILE__), 'tasks/support')
-
-require 'rubygems'
+require 'rake'
 require 'bundler/gem_tasks'
 require 'rspec'
 require 'rspec/core/rake_task'
 
-require 'concurrent'
+require_relative 'lib/extension_helper'
 
 Bundler::GemHelper.install_tasks
 
@@ -16,8 +13,67 @@ Dir.glob('tasks/**/*.rake').each do|rakefile|
   load rakefile
 end
 
+desc "Run benchmarks"
+task :bench do
+  exec "ruby -Ilib -Iext examples/bench_atomic.rb"
+end
+
+if defined?(JRUBY_VERSION)
+  require 'ant'
+
+  EXTENSION_NAME = 'concurrent_jruby'
+
+  directory "pkg/classes"
+
+  desc "Clean up build artifacts"
+  task :clean do
+    rm_rf "pkg/classes"
+    rm_rf "lib/#{EXTENSION_NAME}.jar"
+  end
+
+  desc "Compile the extension"
+  task :compile_java => "pkg/classes" do |t|
+    ant.javac :srcdir => "ext", :destdir => t.prerequisites.first,
+      :source => "1.5", :target => "1.5", :debug => true,
+      :classpath => "${java.class.path}:${sun.boot.class.path}"
+  end
+
+  desc "Build the jar"
+  task :jar => :compile_java do
+    ant.jar :basedir => "pkg/classes", :destfile => "lib/#{EXTENSION_NAME}.jar", :includes => "**/*.class"
+  end
+
+  task :compile => :jar
+
+elsif use_c_extensions?
+
+  EXTENSION_NAME = 'concurrent_cruby'
+
+  require 'rake/extensiontask'
+
+  CLEAN.include Rake::FileList['**/*.so', '**/*.bundle', '**/*.o', '**/mkmf.log', '**/Makefile']
+
+  spec = Gem::Specification.load('concurrent-ruby.gemspec')
+  Rake::ExtensionTask.new(EXTENSION_NAME, spec) do |ext|
+    ext.ext_dir = 'ext'
+    ext.name = EXTENSION_NAME
+    ext.source_pattern = "**/*.{h,c,cpp}"
+  end
+
+  desc 'Clean, compile, and build the extension from scratch'
+  task :compile_c => [ :clean, :compile ]
+
+  task :irb => [:compile] do
+    sh "irb -r ./lib/#{EXTENSION_NAME}.bundle -I #{File.join(File.dirname(__FILE__), 'lib')}"
+  end
+end
+
 RSpec::Core::RakeTask.new(:travis_spec) do |t|
   t.rspec_opts = '--tag ~@not_on_travis'
 end
 
-task :default => [:travis_spec]
+if use_c_extensions?
+  task :default => [:compile_c, :travis_spec]
+else
+  task :default => [:travis_spec]
+end

concurrent-ruby.gemspec

@@ -25,5 +25,12 @@ Gem::Specification.new do |s|
   s.extra_rdoc_files = Dir['README*', 'LICENSE*', 'CHANGELOG*']
   s.require_paths    = ['lib']
 
+  if defined?(JRUBY_VERSION)
+    s.files = Dir['lib/concurrent_jruby.jar']
+    s.platform = 'java'
+  else
+    s.extensions = 'ext/extconf.rb'
+  end
+
   s.required_ruby_version = '>= 1.9.3'
 end

examples/atomic_example.rb

@@ -0,0 +1,12 @@
+require 'concurrent'
+
+my_atomic = Concurrent::Atomic.new(0)
+my_atomic.update {|v| v + 1}
+puts "new value: #{my_atomic.value}"
+
+begin
+  my_atomic.try_update {|v| v + 1}
+rescue Concurrent::Atomic::ConcurrentUpdateError => cue
+  # deal with it (retry, propagate, etc)
+end
+puts "new value: #{my_atomic.value}"

examples/bench_atomic.rb

@@ -0,0 +1,109 @@
+require 'benchmark'
+require 'concurrent'
+require 'thread'
+Thread.abort_on_exception = true
+
+$go = false # for synchronizing parallel threads
+
+# number of updates on the value
+N = ARGV[1] ? ARGV[1].to_i : 100_000
+
+# number of threads for parallel test
+M = ARGV[0] ? ARGV[0].to_i : 100
+
+
+puts "*** Sequential updates ***"
+Benchmark.bm(10) do |x|
+  value = 0
+  x.report "no lock" do
+    N.times do
+      value += 1
+    end
+  end
+
+  @lock = Mutex.new
+  x.report "mutex" do
+    value = 0
+    N.times do
+      @lock.synchronize do
+        value += 1
+      end
+    end
+  end
+
+  @atom = Concurrent::Atomic.new(0)
+  x.report "atomic" do
+    N.times do
+      @atom.update{|x| x += 1}
+    end
+  end
+end
+
+def para_setup(num_threads, count, &block)
+  if num_threads % 2 > 0
+    raise ArgumentError, "num_threads must be a multiple of two"
+  end
+  raise ArgumentError, "need block" unless block_given?
+
+  # Keep those threads together
+  tg = ThreadGroup.new
+
+  num_threads.times do |i|
+    diff = (i % 2 == 0) ? 1 : -1
+
+    t = Thread.new do
+      nil until $go
+      count.times do
+        yield diff
+      end
+    end
+
+    tg.add(t)
+  end
+
+  # Make sure all threads are started
+  while tg.list.find{|t| t.status != "run"}
+    Thread.pass
+  end
+
+  # For good measure
+  GC.start
+
+  tg
+end
+
+def para_run(tg)
+  $go = true
+  tg.list.each{|t| t.join}
+  $go = false
+end
+
+puts "*** Parallel updates ***"
+Benchmark.bm(10) do |bm|
+  # This is not secure
+  value = 0
+  tg = para_setup(M, N/M) do |diff|
+    value += diff
+  end
+  bm.report("no lock"){ para_run(tg) }
+
+
+  value = 0
+  @lock = Mutex.new
+  tg = para_setup(M, N/M) do |diff|
+    @lock.synchronize do
+      value += diff
+    end
+  end
+  bm.report("mutex"){ para_run(tg) }
+  raise unless value == 0
+
+
+  @atom = Concurrent::Atomic.new(0)
+  tg = para_setup(M, N/M) do |diff|
+    @atom.update{|x| x + diff}
+  end
+  bm.report("atomic"){ para_run(tg) }
+  raise unless @atom.value == 0
+
+end

examples/bench_atomic_1.rb

@@ -0,0 +1,138 @@
+#!/usr/bin/env ruby
+
+$: << File.expand_path('../../lib', __FILE__)
+
+require 'optparse'
+require 'thread'
+require 'benchmark'
+
+require 'concurrent'
+
+Thread.abort_on_exception = true
+
+$conf = {
+  :lock => "atomic",
+  :num_threads => 100,
+  :count => 100_000,
+  :count_per_thread => nil,
+  :slow => nil,
+}
+
+OptionParser.new do |opts|
+  opts.on("-c", "--count NUM") do |n|
+    $conf[:count] = n.to_i
+  end
+  opts.on("-p", "--count-per-thread") do |n|
+    $conf[:count_per_thread] = n.to_i
+  end
+  opts.on("-t", "--num-threads NUM") do |n|
+    $conf[:num_threads] = n.to_i
+  end
+  opts.on("-s", "--slow NUM") do |n|
+    $conf[:slow] = n.to_i
+  end
+  opts.on("-l", "--lock atomic|mutex") do |x|
+    $conf[:lock] = x
+  end
+  opts.on("-h", "--help"){ puts opts; exit }
+end.parse!(ARGV)
+
+unless $conf[:count_per_thread]
+  $conf[:count_per_thread] = $conf[:count] / $conf[:num_threads]
+end
+$conf.delete(:count)
+
+if $conf[:slow].to_i > 0
+  require 'digest/md5'
+  def slow_down
+    $conf[:slow].times do |i|
+      Digest::MD5.hexdigest(i.to_s)
+    end
+  end
+
+  ret = []
+  10.times do
+    m = Benchmark.measure{ slow_down }
+    ret << m.real
+  end
+
+  $conf[:slow_time] = [ret.min, ret.max]
+else
+  def slow_down; end
+end
+
+$stderr.puts $conf.inspect
+
+def para_prepare(&block)
+  num_threads = $conf[:num_threads]
+  count = $conf[:count_per_thread]
+
+  if num_threads % 2 > 0
+    raise ArgumentError, "num_threads must be a multiple of two"
+  end
+
+  # Keep those threads together
+  tg = ThreadGroup.new
+
+  num_threads.times do |i|
+    diff = (i % 2 == 0) ? 1 : -1
+
+    t = Thread.new do
+      nil until $go
+      count.times do
+        yield diff
+      end
+    end
+
+    tg.add(t)
+  end
+
+  # Make sure all threads are started
+  while tg.list.find{|t| t.status != "run"}
+    Thread.pass
+  end
+
+  # For good measure
+  GC.start
+
+  $go = false
+
+  tg
+end
+
+
+
+$tg = nil
+if $conf[:lock] == "atomic"
+  $atom = Concurrent::Atomic.new(0)
+  $tg = para_prepare do |diff|
+    $atom.update do |x|
+      slow_down
+      x + diff
+    end
+  end
+else
+  $lock = Mutex.new
+  $value = 0
+  $tg = para_prepare do |diff|
+    $lock.synchronize do
+      slow_down
+      $value += diff
+    end
+  end
+end
+
+
+# Run !
+#
+# NOTE: It seems to me that this measurement method
+#       is sensible to how the system dispatches his resources.
+#
+#       More precise caluclation could be done using
+#       getrusage's times
+ret = Benchmark.measure do
+  $go = true
+  $tg.list.each{|t| t.join}
+  $go = false
+end
+puts ret.real