A FuseFS based RubyGems mirroring spike

rubygems_fuse.rb

require 'fusefs'
require 'net/http/persistent'
require 'dbm'

class Gem::Fuse
  VERSION = '1.0.0'
  SPECZ = "specs.#{Gem.marshal_version}.gz"
  SPECZPATH = '/' + SPECZ
  RUBY = 'ruby'
  REJECT = %r{^/mach_kernel$|^/Backups.backupdb$}
  STAT = %r{/\._[^/]+}
  EMPTY = ''

  def initialize(mount, shadow = nil, db = nil)
    @http = Net::HTTP::Persistent.new
    @root = URI 'http://production.cf.rubygems.org/'
    db ||= File.expand_path('../gem-fuse', mount)
    @fsdb = DBM.open(db)

    @shadow = shadow
    @queue = Queue.new
    @worker = shadow_worker

    FuseFS.set_root( self )
    FuseFS.mount_under mount
  end

  def stop
    @fsdb.close
    FuseFS.exit
    FuseFS.unmount
  end

  def run
    FuseFS.run
  end

  def file?(path)
    return false if path =~ REJECT
    return true if directory?(path)
    uri = uri path
    @fsdb[uri.path] || begin
      r = raw_request uri, head(uri)
      r.code.to_i == 200
    end
  end

  def size(path)
    return false if path =~ REJECT
    uri = uri path
    @fsdb.delete(SPECZPATH) if uri.path == SPECZPATH
    size = @fsdb[uri.path] ||= begin
      r = raw_request uri, head(uri)
      r.content_length
    end
    size.to_i
  end

  def directory?(path)
    path = uri(path).path # N.B. runs the stat file cleanup
    path == '/gems' || path == '/'
  end

  def contents(path)
    case path
    when '/gems'
      gems = request @root + SPECZ
      gems = Gem.gunzip gems
      gems = Marshal.load gems
      gems.map! { |n,v,p| "#{n}-#{v}#{"-#{p}" unless p == RUBY}.gem" }
    when '/'
      [SPECZ, 'gems']
    else
      []
    end
  end

  def raw_open(path, mode)
    mode == 'r' && file?(path)
  end

  def raw_read(path, offset, size)
    if file = shadow_file(path)
      File.open(file, 'rb') do |f|
        f.scan(offset)
        f.read(size)
      end
    else
      uri = uri path
      request uri, range(uri, offset, size)
    end
  end

  def raw_close(path)
    true
  end
  
  def shutdown
    @queue.push nil # tells the worker to stop
    @worker.join
  end

  private
  def uri(path)
    @root + path.sub(STAT, EMPTY)
  end

  def head(uri)
    Net::HTTP::Head.new(uri.path)
  end

  def range(uri, offset, size)
    r = Net::HTTP::Get.new(uri.path)
    r.set_range offset, size
    r
  end

  def raw_request(uri, req = nil, &b)
    p [uri.path, req]
    @http.request(uri, req, &b)
  rescue
    p $!, *$@
  end

  def request(uri, req = nil, &b)
    r = raw_request uri, req, &b
    case r
    when Net::HTTPSuccess
      b ? r : r.body
    else
      p [uri.path, :failed!]
      nil
    end
  end

  # Only return a path if it's likely to be valid.
  def shadow_file(path)
    file = File.expand_path(path, @shadow)

    # If the file doesn't exist, enqueue it.
    unless File.exists?(file)
      @queue << path
      return
    end

    if File.size(file) == size(file)
      p [file, :read_shadowed]
      file
    else
      p [file, :shadow_size_mismatch]
      nil
    end
  end

  def shadow_worker
    Thread.new do
      loop do
        while path = @queue.pop
          uri = uri(path)
          request uri do |res|
            shadow_path = File.expand_path(path, @shadow)
            p [:shadowing, uri.path]
            open(shadow_path) do |f|
              res.read_body do |chunk|
                f << chunk
              end
            end
          end
        end
      end
    end
  end
end

BEGIN { require 'rubygems' if __FILE__ == $0 }
if __FILE__ == $0
  mount = ARGV.shift
  shadow = ARGV.shift

  unless mount && shadow
    abort "Usage: #{File.basename($0)} mount_path shadow_path"
  end

  begin
    fuse = Gem::Fuse.new mount
    trap(:INT) { fuse.stop }
    fuse.run
  ensure
    fuse.shutdown
  end
end