tqueue.rb

require 'concurrent'

class TBQueue
  def initialize size
    tvar = Concurrent::TVar

    @read = tvar.new []
    @write = tvar.new []
    @rsize = tvar.new 0
    @wsize = tvar.new size
  end

  def write(a)
    w = t.read(@wsize)

    if w != 0
      t.write(@wsize, w - 1)
    else
      r = t.read(@rsize)
      if r != 0
        t.write(@rsize, 0)
        t.write(@wsize, r - 1)
      else
        retry_transaction
      end
    end

    as = t.read(@write)
    as.push(a)
    t.write(@write, as)

    return nil
  end

  def read
    xs = t.read(@read)
    r = t.read(@rsize)
    t.write(@rsize, r + 1)

    if !xs.empty?
      x, *xs_ = xs
      t.write(@read, xs_)
      return x
    else
      ys = t.read(@write)
      if ys.empty?
        retry_transaction
      else
        y, *ys_ = ys
        t.write(@write, [])
        t.write(@read, ys_)

        return y
      end
    end
  end

  def peek
    x = read
    unshift x

    return x
  end

  def unget a
    r = t.read @rsize
    if r > 0
      t.write @rsize, r - 1
    else
      w = t.read @wsize
      if w > 0
        t.write @wsize, w - 1
      else
        retry_transaction
      end
    end

    xs = t.read @read
    t.write @read, xs.unshift(a)

    return nil
  end

  private
  def t
    Concurrent::Transaction.current
  end

  def retry_transaction
    Concurrent.abort_transaction
  end
end