Code examples from my "Using threads in Ruby applications" presentation in our local Ruby meetup http://www.meetup.com/ruby-on-rails-latvia/events/105142132/

01_thread.rb

Thread.new do
  sleep 5
  puts "finished"
end

thread = Thread.new do
  sleep 5
  puts "finished"
end
thread.join

thread = Thread.new(5) do |seconds|
  sleep seconds
  puts "finished in #{seconds} seconds"
end
thread.join

thread = Thread.new do
  raise "exception"
end
thread.join

# false by default and better leave it to false
Thread.abort_on_exception = true

02_web_requests.rb

require "rubygems"
require "faraday"
require "json"

conn = Faraday.new url: 'https://jira.atlassian.com'

per_page = 100

10.times do |i|
  puts "Request #{i + 1}"
  response = conn.get "/rest/api/latest/search", jql: "project=JIRA",
    startAt: i * per_page, maxResults: per_page
  json = JSON.parse response.body
  json['issues'].each do |issue|
    puts "#{issue['key']}: #{issue['fields']['summary']}"
  end
end

03_web_requests_threaded.rb

require "rubygems"
require "faraday"
require "json"

conn = Faraday.new url: 'https://jira.atlassian.com'

per_page = 100

threads = Array.new(10) do |i|
  Thread.new(i) do
    puts "Request #{i + 1}"
    response = conn.get "/rest/api/latest/search", jql: "project=JIRA",
      startAt: i * per_page, maxResults: per_page
    json = JSON.parse response.body
    json['issues'].map do |issue|
      "#{issue['key']}: #{issue['fields']['summary']}"
    end
  end
end

threads.each do |thread|
  puts thread.value.join("\n")
end

04_web_requests_thread_pool.rb

require "rubygems"
require "faraday"
require "json"
require "thread"

conn = Faraday.new url: 'https://jira.atlassian.com'

total = 5000
per_page = 100

offset_queue = Queue.new
(total / per_page + 1).times{|i| offset_queue.push i * per_page}

(pool_size = 10).times{ offset_queue.push :END_OF_WORK}

results_queue = Queue.new

threads = Array.new(pool_size) do
  Thread.new do
    while offset = offset_queue.pop
      break if offset == :END_OF_WORK
      puts "Request offset #{offset}"
      response = conn.get "/rest/api/latest/search", jql: "project=JIRA", startAt: offset, maxResults: per_page
      json = JSON.parse response.body
      results = Array(json['issues']).map do |issue|
        "#{issue['key']}: #{issue['fields']['summary']}"
      end
      results_queue.push results
    end
  end
end

threads.each(&:join)

all_results = []
while !results_queue.empty?
  all_results.concat results_queue.pop
end
puts "all_results.size=#{all_results.size}"

05_threaded_counter.rb

require "thread"

count = 0

threads = Array.new(10) do
  Thread.new do
    100_000.times do |i|
      count += 1
    end
  end
end

threads.each(&:join)

# will be 1_000_000 on MRI and non-deterministic on JRuby
puts "count=#{count}"

05_threaded_counter2.rb

require "thread"

count = 0

threads = Array.new(10) do
  Thread.new do
    100_000.times do |i|
      new_value = count + 1
      print "#{new_value}\n" if new_value % 250_000 == 0
      count = new_value
    end
  end
end

threads.each(&:join)

# will be non-deterministic on MRI and JRuby
# as thread switching will happen also on MRI during print I/O
puts "count=#{count}"

05_threaded_counter3.rb

count = 0

mutex = Mutex.new

threads = Array.new(10) do
  Thread.new do
    100_000.times do |i|
      mutex.synchronize do
        count += 1
      end
    end
  end
end

threads.each(&:join)

# will be 1_000_000 on both MRI and JRuby
puts "count=#{count}"

06_atomic_counter.rb

require "rubygems"
require "atomic"

count = Atomic.new(0)

threads = Array.new(10) do
  Thread.new do
    100_000.times do |i|
      count.update{|value| value + 1}
    end
  end
end

threads.each(&:join)

# will be 1_000_000 on both MRI and JRuby
puts "count=#{count.value}"

06_atomic_counter_benchmark.rb

require "rubygems"
require "atomic"
require "benchmark"

Benchmark.bm do |x|
  5.times do |i|
    x.report "with mutex #{i}" do
      count = 0
      mutex = Mutex.new
      threads = Array.new(10) do
        Thread.new do
          100_000.times do |i|
            mutex.synchronize do
              count += 1
            end
          end
        end
      end
      threads.each(&:join)
      # puts "count=#{count}"
    end
  end

end

Benchmark.bm do |x|
  5.times do |i|
    x.report "with atomic #{i}" do
      count = Atomic.new(0)
      threads = Array.new(10) do
        Thread.new do
          100_000.times do |i|
            count.update{|value| value + 1}
          end
        end
      end
      threads.each(&:join)
      # puts "count=#{count.value}"
    end
  end
end

06_thread_safe.rb

require "rubygems"
require "thread_safe"

hash = {}
hash[:a] ||= begin
  # complex calculation
  sleep 1
  1
end

# used in Rails 4 in many places instead of previous Hash
cache = ThreadSafe::Cache.new
# thread safe version
cache[:a] ||= begin
  # complex calculation
  sleep 1
  1
end

07_thread_variables.rb

count = 0

threads = Array.new(10) do |i|
  Thread.new do
    sleep rand(0.1)
    Thread.current[:count] = count
    count += 1
  end
end

threads.each {|t| t.join; print t[:count], ", " }
puts "count = #{count}"

08_class_attributes.rb

class User
  def self.current
    @current
  end
  def self.current(current_user)
    @current = current_user
  end
end

class ApplicationController
  before_filter :set_current_user

  private
  def set_current_user
    User.current = current_user
  end
end


# in threaded application server
# replace with
class User
  def self.current
    Thread.current[:user_current]
  end
  def self.current(current_user)
    Thread.current[:user_current] = current_user
  end
end

09_fiber.rb

fibonacci = Fiber.new do
  Fiber.yield (f1 = 1)
  Fiber.yield (f2 = 2)
  while true
    f1, f2 = f2, f1 + f2
    Fiber.yield f2
  end
end

20.times do
  puts fibonacci.resume
end

09_fiber_thread_variables.rb

Thread.new do
  puts "in thread Thread.current.object_id=#{Thread.current.object_id}"
  Thread.current[:a] = 1
  puts "in thread Thread.current[:a]=#{Thread.current[:a].inspect}"
  Fiber.new do
    puts "in fiber Thread.current.object_id=#{Thread.current.object_id}"
    # will be nil on MRI and 1 on JRuby 1.7.3
    puts "in fiber Thread.current[:a]=#{Thread.current[:a].inspect}"
    Thread.current[:a] = 2
    puts "in fiber Thread.current[:a]=#{Thread.current[:a].inspect}"
  end.resume

  # will be 1 on MRI and 2 on JRuby 1.7.3
  puts "in thread Thread.current[:a]=#{Thread.current[:a].inspect}"
end.join

10_activerecord_connection_pool.rb

require "rubygems"
require "active_record"
require "mysql2"

ActiveRecord::Base.establish_connection(
  adapter: "mysql2",
  database: "thread_test",
  encoding: "utf8",
  pool: 5,
  username: "root"
)

ActiveRecord::Base.connection.create_table "dummies", force: true do |t|
  t.string :dummy
end

class Dummy < ActiveRecord::Base
end

# This will fail due to timeout on connection pool which size is 5
# but 10 threads want to get their connections
Array.new(10) do
  Thread.new do
    print "Dummy.connection.object_id=#{Dummy.connection.object_id}\n"
    1_000.times do
      Dummy.create! dummy: "dummy"
    end
  end
end.each(&:join)


# This will succeed even when database connection pool size is 5
# but threads are 10.
# It is recommended always to do connection_pool.with_connection
# which ensures that database connections are checked in back to pool.
Array.new(10) do
  Thread.new do
    1_000.times do
      Dummy.connection_pool.with_connection do |connection|
        Dummy.create! dummy: "dummy"
      end
    end
  end
end.each(&:join)