lib/mogilefs/http_file.rb

   1 # -*- encoding: binary -*-
   2 # here are internal implementation details, do not use them in your code
   3 require 'stringio'
   4 require 'uri'
   5 require 'digest/md5'
   6 require 'mogilefs/chunker'
   7
   8 ##
   9 # HTTPFile wraps up the new file operations for storing files onto an HTTP
  10 # storage node.
  11 #
  12 # You really don't want to create an HTTPFile by hand.  Instead you want to
  13 # create a new file using MogileFS::MogileFS.new_file.
  14 #
  15 class MogileFS::HTTPFile < StringIO
  16   class RetryableError < MogileFS::Error; end
  17   class EmptyResponseError < RetryableError; end
  18   class BadResponseError < RetryableError; end
  19   class UnparseableResponseError < RetryableError; end
  20   class NoStorageNodesError < MogileFS::Error
  21     def message; 'Unable to open socket to storage node'; end
  22   end
  23   class NonRetryableError < MogileFS::Error; end
  24
  25   class HTTPSock < MogileFS::Socket
  26     attr_accessor :start
  27
  28     # Increase timeout as we become more invested in uploading with
  29     # this socket.  The server could be experiencing I/O delays
  30     # from large uploads because the sysadmin forgot to tune the
  31     # VM sysctls for handling large files.
  32     def write(buf)
  33       timed_write(buf, Time.now - @start + 5.0)
  34     end
  35   end
  36
  37   # :stopdoc:
  38   MD5_TRAILER_NODES = {} # :nodoc: # EXPERIMENTAL
  39   class << self
  40     attr_accessor :response_timeout_cb
  41   end
  42
  43   # temporary directories (nginx) may not be configured on the
  44   # same device, necessitating a time-consuming full file copy
  45   # instead of a quick rename(2)/link(2) operation
  46   @response_timeout_cb = lambda do |elapsed_time, bytes_uploaded|
  47     mbytes_uploaded = bytes_uploaded / (1024.0 * 1024.0)
  48     # assumes worst case is 10M/s on the remote storage disk
  49     t = mbytes_uploaded * 10 + elapsed_time
  50     t < 5 ? 5 : t
  51   end
  52   # :startdoc:
  53
  54   ##
  55   # The big_io name in case we have file > 256M
  56
  57   attr_accessor :big_io
  58
  59   attr_accessor :streaming_io
  60
  61   ##
  62   # Creates a new HTTPFile with MogileFS-specific data.  Use
  63   # MogileFS::MogileFS#new_file instead of this method.
  64
  65   def initialize(dests, opts = nil)
  66     super ""
  67     @md5 = @streaming_io = @big_io = @active = nil
  68     @dests = dests
  69     @opts = Integer === opts ? { :content_length => opts } : opts
  70   end
  71
  72   def request_put(sock, uri, file_size, input = nil)
  73     host_with_port = "#{uri.host}:#{uri.port}"
  74     clen = @opts[:content_length]
  75     file_size ||= clen
  76
  77     content_md5 = @opts[:content_md5]
  78     if String === content_md5
  79       file_size or
  80         raise ArgumentError,
  81               ":content_length must be specified with :content_md5 String"
  82       file_size = "#{file_size}\r\nContent-MD5: #{content_md5}"
  83     elsif content_md5.respond_to?(:call) ||
  84           :trailer == content_md5 ||
  85           MD5_TRAILER_NODES[host_with_port]
  86       file_size = nil
  87       @md5 = Digest::MD5.new
  88     end
  89
  90     if file_size
  91       sock.write("PUT #{uri.request_uri} HTTP/1.0\r\n" \
  92                  "Content-Length: #{file_size}\r\n\r\n")
  93       rv = input ? MogileFS.io.copy_stream(@active = input, sock) : yield(sock)
  94     else
  95       trailers = @md5 ? "Trailer: Content-MD5\r\n" : ""
  96       sock.write("PUT #{uri.request_uri} HTTP/1.1\r\n" \
  97                  "Host: #{host_with_port}\r\n#{trailers}" \
  98                  "Transfer-Encoding: chunked\r\n\r\n")
  99       tmp = MogileFS::Chunker.new(sock, @md5, content_md5)
 100       rv = input ? MogileFS.io.copy_stream(@active = input, tmp) : yield(tmp)
 101       tmp.flush
 102     end
 103
 104     if clen && clen != rv
 105       raise MogileFS::SizeMismatchError,
 106             ":content_length expected: #{clen.inspect}, actual: #{rv.inspect}"
 107     end
 108     rv
 109   end
 110
 111   def put_streaming_io(sock, uri) # unlikely to be used
 112     file_size = @streaming_io.length
 113     written = 0
 114     request_put(sock, uri, file_size) do |wr|
 115       @streaming_io.call(Proc.new do |data_to_write|
 116         written += wr.write(data_to_write)
 117       end)
 118     end
 119     file_size ? file_size : written
 120   end
 121
 122   def rewind_or_raise!(uri, err)
 123     @active.rewind if @active
 124     rescue => e
 125       msg = "#{uri} failed with #{err.message} (#{err.class}) and " \
 126             "retrying is impossible as rewind on " \
 127             "#{@active.inspect} failed with: #{e.message} (#{e.class})"
 128       raise NonRetryableError, msg, e.backtrace
 129   end
 130
 131   ##
 132   # Writes an HTTP PUT request to +sock+ to upload the file and
 133   # returns file size if the socket finished writing
 134   def upload(devid, uri) # :nodoc:
 135     start = Time.now
 136     sock = HTTPSock.tcp(uri.host, uri.port)
 137     sock.start = start
 138     file_size = length
 139
 140     if @streaming_io
 141       file_size = put_streaming_io(sock, uri)
 142     elsif @big_io
 143       if String === @big_io || @big_io.respond_to?(:to_path)
 144         file = File.open(@big_io)
 145         stat = file.stat
 146         file_size = request_put(sock, uri, stat.file? ? stat.size : nil, file)
 147       else
 148         size = nil
 149         if @big_io.respond_to?(:stat)
 150           stat = @big_io.stat
 151           size = stat.size if stat.file?
 152         elsif @big_io.respond_to?(:size)
 153           size = @big_io.size
 154         end
 155         file_size = request_put(sock, uri, size, @big_io)
 156       end
 157     else
 158       rewind
 159       request_put(sock, uri, file_size, self)
 160     end
 161
 162     tout = self.class.response_timeout_cb.call(Time.now - start, file_size)
 163
 164     case line = sock.timed_read(23, "", tout)
 165     when %r{^HTTP/\d\.\d\s+(2\d\d)\s} # success!
 166       file_size
 167     when nil
 168       raise EmptyResponseError, 'Unable to read response line from server'
 169     when %r{^HTTP/\d\.\d\s+(\d+)}
 170       raise BadResponseError, "HTTP response status from upload: #$1"
 171     else
 172       raise UnparseableResponseError,
 173             "Response line not understood: #{line.inspect}"
 174     end
 175     rescue SystemCallError, RetryableError => err
 176       rewind_or_raise!(uri, err)
 177       raise
 178     ensure
 179       file.close if file
 180       sock.close if sock
 181   end
 182
 183   def commit
 184     errors = nil
 185     @dests.each do |devid, path|
 186       begin
 187         uri = URI.parse(path)
 188         bytes_uploaded = upload(devid, uri)
 189         return create_close(devid, uri, bytes_uploaded)
 190       rescue SystemCallError, RetryableError => e
 191         errors ||= []
 192         errors << "#{path} - #{e.message} (#{e.class})"
 193       end
 194     end
 195
 196     raise NoStorageNodesError,
 197           "all paths failed with PUT: #{errors.join(', ')}", []
 198   end
 199
 200   def create_close(devid, uri, bytes_uploaded)
 201     args = {
 202       :fid => @opts[:fid],
 203       :devid => devid,
 204       :key => @opts[:key],
 205       :domain => @opts[:domain],
 206       :size => bytes_uploaded,
 207       :path => uri.to_s,
 208     }
 209     if @md5
 210       args[:checksum] = "MD5:#{@md5.hexdigest}"
 211     elsif String === @opts[:content_md5]
 212       hex = @opts[:content_md5].unpack('m')[0].unpack('H*')[0]
 213       args[:checksum] = "MD5:#{hex}"
 214     end
 215     args[:checksumverify] = 1 if @opts[:checksumverify]
 216     @opts[:backend].create_close(args)
 217     bytes_uploaded
 218   end
 219
 220   def close
 221     commit
 222     super
 223   end
 224 end