mogstored_rack: add ability to change :io_size and :open_flags

This allows folks to try IO::SYNC, and maybe even IO::DIRECT.
Additionally, :io_size may be used to optimize IO::SYNC or
make IO::DIRECT _work_.

Removal of the "Tempfile" dependency means we no longer work
reliably on NFS, but nobody sane puts MogileFS devices on
NFS, anyways.  (Actually, nobody sane uses NFS if atomicity
is important :P)

1 files changed, 35 insertions, 36 deletions

diff --git a/examples/mogstored_rack.rb b/examples/mogstored_rack.rb
index caa779a..7ac90d6 100644
--- a/examples/mogstored_rack.rb
+++ b/examples/mogstored_rack.rb
@@ -1,5 +1,4 @@
 # -*- encoding: binary -*-
-require 'tempfile'
 require 'digest/md5'
 require 'rack'
 
@@ -22,11 +21,28 @@ class MogstoredRack
 
   def initialize(root, opts = {})
     @root = File.expand_path(root)
+    @io_size = opts[:io_size] || 0x100000
     @rack_file = (opts[:app] || Rack::File.new(@root))
     @fsync = !! opts[:fsync]
     @creat_perms = opts[:creat_perms] || (~File.umask & 0666)
     @mkdir_perms = opts[:mkdir_perms] || (~File.umask & 0777)
     @reread_verify = !! opts[:reread_verify]
+    @open_flags = opts[:open_flags] || 0
+    @open_flags |= IO::RDWR | IO::CREAT | IO::EXCL
+  end
+
+  def tmpfile(basename, dir)
+    t = Time.now.utc.strftime("%Y%m%d%H%M%S")
+    seq = 0
+    begin
+      fp = File.open("#{dir}/#{basename}.#{t}.#{seq}.tmp", @open_flags, 0600)
+    rescue Errno::EEXIST
+      retry if (seq += 1) < 10
+      raise
+    end
+    fp.binmode
+    fp.sync = true
+    fp
   end
 
   def call(env)
@@ -75,32 +91,29 @@ class MogstoredRack
     dir = File.dirname(path)
     File.directory?(dir) or return r(403)
 
-    Tempfile.open([File.basename(path), ".tmp"], dir) do |tmp|
-      tmp = tmp.to_io # delegated method calls are slower
-      tmp.sync = true
-      tmp.binmode
-      buf = ""
-      received = put_loop(env["rack.input"], tmp, buf)
-      err = content_md5_fail?(env, received) and return err
-      if @reread_verify && err = reread_md5_fail?(env, tmp, received, buf)
-        return err
-      end
-      tmp.chmod(@creat_perms)
-      begin
-        File.link(tmp.path, path)
-      rescue Errno::EEXIST
-        err = rename_overwrite_fail?(tmp.path, path) and return err
-      end
-      fsync(dir, tmp) if @fsync
-      resp = r(201)
-      resp[1]["X-Received-Content-MD5"] = received
-      return resp
+    tmp = tmpfile(File.basename(path), dir)
+    buf = ""
+    received = put_loop(env["rack.input"], tmp, buf)
+    err = content_md5_fail?(env, received) and return err
+    if @reread_verify && err = reread_md5_fail?(env, tmp, received, buf)
+      return err
     end
+    tmp.chmod(@creat_perms)
+    File.rename(tmp.path, path)
+    fsync(dir, tmp) if @fsync
+    resp = r(201)
+    resp[1]["X-Received-Content-MD5"] = received
+    resp
+    rescue
+      File.unlink(tmp.path) if tmp && File.exist?(tmp.path)
+      raise
+    ensure
+      tmp.close if tmp
   end
 
   def put_loop(src, dst, buf)
     md5 = ContentMD5.new
-    while src.read(0x4000, buf)
+    while src.read(@io_size, buf)
       md5.update(buf)
       dst.write(buf)
     end
@@ -167,20 +180,6 @@ class MogstoredRack
            "received: #{received}", env)
   end
 
-  def rename_overwrite_fail?(src, dst)
-    10.times do
-      begin
-        tmp_dst = "#{dst}.#{rand}"
-        File.link(src, tmp_dst)
-      rescue Errno::EEXIST
-        next
-      end
-      File.rename(tmp_dst, dst)
-      return false # success!
-    end
-    r(409)
-  end
-
   # fsync each and every directory component above us on the same device
   def fsync(dir, tmp)
     tmp.fsync