lib/mogilefs/new_file/common.rb

   1 # -*- encoding: binary -*-
   2 # here are internal implementation details, do not use them in your code
   3 require 'socket'
   4 require 'uri'
   5 require 'digest/md5'
   6 require 'mogilefs/chunker'
   7
   8 module MogileFS::NewFile::Common
   9   # :stopdoc:
  10   class RetryableError < MogileFS::Error; end
  11   class EmptyResponseError < RetryableError; end
  12   class BadResponseError < RetryableError; end
  13   class UnparseableResponseError < RetryableError; end
  14   class NoStorageNodesError < MogileFS::Error
  15     def message; 'Unable to open socket to storage node'; end
  16   end
  17   class NonRetryableError < MogileFS::Error; end
  18
  19   MD5_TRAILER_NODES = {} # :nodoc: # EXPERIMENTAL
  20
  21   def read_response(sock)
  22     # mostly relying on SO_KEEPALIVE to timeout
  23     case line = sock.timed_read(23, "", 7200)
  24     when %r{^HTTP/\d\.\d\s+(2\d\d)\s} # success!
  25     when nil
  26       raise EmptyResponseError, 'Unable to read response line from server'
  27     when %r{^HTTP/\d\.\d\s+(\d+)}
  28       raise BadResponseError, "HTTP response status from upload: #$1"
  29     else
  30       raise UnparseableResponseError,
  31             "Response line not understood: #{line.inspect}"
  32     end
  33   end
  34
  35   def create_close(devid, uri, bytes_uploaded)
  36     args = {
  37       :fid => @opts[:fid],
  38       :devid => devid,
  39       :key => @opts[:key],
  40       :domain => @opts[:domain],
  41       :size => bytes_uploaded,
  42       :path => uri.to_s,
  43     }
  44     if @md5
  45       args[:checksum] = "MD5:#{@md5.hexdigest}"
  46     elsif String === @opts[:content_md5]
  47       hex = @opts[:content_md5].unpack('m')[0].unpack('H*')[0]
  48       args[:checksum] = "MD5:#{hex}"
  49     end
  50     args[:checksumverify] = 1 if @opts[:checksumverify]
  51     backend = @opts[:backend]
  52
  53     # upload could've taken a long time, ping and try to ensure socket
  54     # is valid to minimize (but not completely eliminate) the chance
  55     # create_close hits a stale socket (while reading the response after
  56     # writing to it) and becomes non-retryable.  We treat create_close
  57     # specially as its less idempotent than any other command
  58     # (even other non-idempotent ones).  There may be no hope of retrying
  59     # the upload at all if data was streamed and calling create_close
  60     # twice will hurt us...
  61     backend.noop
  62
  63     backend.create_close(args)
  64     bytes_uploaded
  65   end
  66
  67   # aggressive keepalive settings on Linux + Ruby 1.9.2+
  68   TCP_KEEPALIVE = {
  69     :TCP_KEEPIDLE => 60, # seconds time before keepalive packet is sent
  70     :TCP_KEEPINTVL => 5,
  71     :TCP_KEEPCNT => 2,  # number of retries
  72   }
  73
  74   req_consts = TCP_KEEPALIVE.keys
  75   if (Socket.constants & req_consts).size == req_consts.size
  76     def set_socket_options(sock)
  77       sock.setsockopt(:SOL_SOCKET, :SO_KEEPALIVE, 1)
  78       TCP_KEEPALIVE.each do |k,v|
  79         sock.setsockopt(:IPPROTO_TCP, k, v)
  80       end
  81     end
  82   else
  83     def set_socket_options(sock)
  84       sock.setsockopt(Socket::SOL_SOCKET, Socket::SO_KEEPALIVE, 1)
  85     end
  86   end
  87   # :startdoc:
  88 end