Cacher is the code snippet organizer for pro developers

require 'thread'

class TwoLockLinkedQueue
  # This Node doesn't need atomic updates, it assumes
  # that you're modifying it while holding a lock
  Node = Struct.new(:item, :successor)

  def initialize
    dummy_node = Node.new(:dummy, nil)

    @head_node = dummy_node
    @tail_node = dummy_node

    @head_lock = Mutex.new
    @tail_lock = Mutex.new
  end

  def push(item)
    # allocate a new node with the item embedded
    new_node = Node.new(item, nil)

    @tail_lock.synchronize do
      # update the successor of the current tail to point to the new node
      @tail_node.successor = new_node
      @tail_node = new_node
    end
  end

  def pop
    @head_lock.synchronize do
      dummy_node = @head_node
      head = @head_node.successor

      if head.nil? # then queue was empty
        return nil
      else
        # the current head becomes the new 'dummy' head
        @head_node = head
        # return its value
        return head.item
      end
    end
  end

  def size
    successor = @head_node.successor
    count = 0

    loop do
      break if successor.nil?

      current_node = successor
      successor = current_node.successor
      count += 1
    end

    count
  end
end

require 'benchmark'
require 'thread'
require_relative 'atomic_linked_queue'
require_relative 'two_lock_linked_queue'

thread_count = 50

queue_klasses = [Queue, TwoLockLinkedQueue, AtomicLinkedQueue]
Thread.abort_on_exception = true

# this one tells all the threads when to start
$go = false

def setup(queue, thread_count, queue_length)
  tg = ThreadGroup.new

  if queue.class == Queue
    thread_count.times do
      t = Thread.new do
        # wait until the bm starts to do the work. This should
        # minimize variance.
        nil until $go
        (queue_length / thread_count).to_i.times do
          queue.push('item')
        end

        loop do
          begin
            result = queue.pop(:nonblock)
          rescue => e
            break
          end
        end
      end

      tg.add(t)
    end

  else
    thread_count.times do
      t = Thread.new do
        nil until $go
        (queue_length / thread_count).to_i.times do
          queue.push('item')
        end

        result = true
        until result.nil?
          result = queue.pop
        end
      end

      tg.add(t)
    end
  end

  GC.start
  tg
end

def exercise(tg)
  $go = true
  tg.list.each(&:join)
  $go = false
end

3.times do
  queue_klasses.each do |klass|
    Benchmark.bm(45) do |bm|
      queue = klass.new
      tg = setup(queue, thread_count, 10_000)
      bm.report("#{klass} - fill, then empty - 10k") { exercise(tg) }
      raise 'hell' unless queue.size.zero?
      
      queue = klass.new
      tg = setup(queue, thread_count, 100_000)
      bm.report("#{klass} - fill, then empty - 100k") { exercise(tg) }
      raise 'hell' unless queue.size.zero?

      queue = klass.new
      tg = setup(queue, thread_count, 1_000_000)
      bm.report("#{klass} - fill, then empty - 1mm") { exercise(tg) }
      raise 'hell' unless queue.size.zero?
    end
  end
end

require 'benchmark'
require 'thread'
require_relative 'atomic_linked_queue'
require_relative 'two_lock_linked_queue'

thread_count = 50
iterations = 10_000

queue_klasses = [Queue, TwoLockLinkedQueue, AtomicLinkedQueue]
Thread.abort_on_exception = true

# this one tells all the threads when to start
$go = false

def setup(queue, writer_thread_count, reader_thread_count, iterations)
  tg = ThreadGroup.new
  
  # spawn writer threads
  writer_thread_count.times do
    t = Thread.new do
      # wait until the bm starts to do the work. This should
      # minimize variance.
      nil until $go
      iterations.times do
        queue.push('item')
      end
    end

    tg.add(t)
  end
  
  # spawn reader threads
  if queue.class == Queue
    # the Queue class gets special behaviour because its #pop
    # method is blocking by default.
    reader_thread_count.times do
      t = Thread.new do
        nil until $go
        iterations.times do
          begin
            queue.pop(:nonblocking)
          rescue
          end
        end
      end

      tg.add(t)
    end
  else
    reader_thread_count.times do
      t = Thread.new do
        nil until $go
        iterations.times do
          queue.pop
        end
      end

      tg.add(t)
    end
  end
  
  tg
end

def exercise(tg)
  $go = true
  tg.list.each(&:join)
  $go = false
end

3.times do
  queue_klasses.each do |klass|
    Benchmark.bm(50) do |bm|
      queue = klass.new
      tg = setup(queue, thread_count, (thread_count * 0.6).to_i, iterations)
      bm.report("#{klass} - more writers than readers") { exercise(tg) }

      queue = klass.new
      tg = setup(queue, (thread_count * 0.6).to_i, thread_count, iterations)
      bm.report("#{klass} - more readers than writers") { exercise(tg) }

      queue = klass.new
      tg = setup(queue, thread_count, thread_count, iterations)
      bm.report("#{klass} - equal writers and readers") { exercise(tg) }
    end
  end
end

# This is a proof-of-concept of a concurrent, lock-free FIFO data 
# structure written in Ruby. It leverages atomic updates, rather than
# lock-based synchronization.

require 'atomic'

class AtomicLinkedQueue
  class Node
    attr_accessor :item

    def initialize(item, successor)
      @item = item
      @successor = Atomic.new(successor)
    end

    def successor
      @successor.value
    end

    def update_successor(old, new)
      @successor.compare_and_set(old, new)
    end
  end

  def initialize
    dummy_node = Node.new(:dummy, nil)

    @head = Atomic.new(dummy_node)
    @tail = Atomic.new(dummy_node)
  end

  def push(item)
    # allocate a new node with the item embedded
    new_node = Node.new(item, nil)

    # keep trying until the operation succeeds
    loop do
      current_tail_node = @tail.value
      current_tail_successor = current_tail_node.successor

      # if our stored tail is still the current tail
      if current_tail_node == @tail.value
        # if that tail was really the last node
        if current_tail_successor.nil?
          # if we can update the previous successor of tail to point to this new node
          if current_tail_node.update_successor(nil, new_node)
            # then update tail to point to this node as well
            @tail.compare_and_set(current_tail_node, new_node)
            # and return
            return true
            # else, start the loop over
          end
        else
          # in this case, the tail ref we had wasn't the real tail
          # so we try to set its successor as the real tail, then start the loop again
          @tail.compare_and_set(current_tail_node, current_tail_successor)
        end
      end
    end
  end

  def pop
    # retry until some value can be returned
    loop do
      # the value in @head is just a dummy node that always sits in that position,
      # the real 'head' is in its successor
      current_dummy_node = @head.value
      current_tail_node = @tail.value

      current_head_node = current_dummy_node.successor

      # if our local head is still consistent with the head node, continue
      # otherwise, start over
      if current_dummy_node == @head.value
        # if either the queue is empty, or falling behind
        if current_dummy_node == current_tail_node
          # if there's nothing after the 'dummy' head node
          if current_head_node.nil?
            # just return nil
            return nil
          else
            # here the head element succeeding head is not nil, but the head and tail are equal
            # so tail is falling behind, update it, then start over
            @tail.compare_and_set(current_tail_node, current_head_node)
          end
        # the queue isn't empty
        # if we can set the dummy head to the 'real' head, we're free to return the value in that real head, success
        elsif @head.compare_and_set(current_dummy_node, current_head_node)
          # grab the item from the popped node
          item = current_head_node.item

          if item != nil
            current_head_node.item = nil
          end
          
          # return it, success!
          return item

        # else
          # try again
        end
      end
    end
  end

  def size
    successor = @head.value.successor
    count = 0

    loop do
      break if successor.nil?

      current_node = successor
      successor = current_node.successor
      count += 1
    end

    count
  end
end