lib/mogilefs/backend.rb

   1 # -*- encoding: binary -*-
   2 require 'thread'
   3
   4 ##
   5 # MogileFS::Backend communicates with the MogileFS trackers.
   6
   7 class MogileFS::Backend
   8
   9   ##
  10   # Adds MogileFS commands +names+.
  11
  12   def self.add_command(*names)
  13     names.each do |name|
  14       define_method name do |*args|
  15         do_request(name, args[0] || {}, false)
  16       end
  17     end
  18   end
  19
  20   def self.add_idempotent_command(*names)
  21     names.each do |name|
  22       define_method name do |*args|
  23         do_request(name, args[0] || {}, true)
  24       end
  25     end
  26   end
  27
  28   BACKEND_ERRORS = {}
  29
  30   # this converts an error code from a mogilefsd tracker to an exception:
  31   #
  32   # Examples of some exceptions that get created:
  33   #   class AfterMismatchError < MogileFS::Error; end
  34   #   class DomainNotFoundError < MogileFS::Error; end
  35   #   class InvalidCharsError < MogileFS::Error; end
  36   def self.add_error(err_snake)
  37     err_camel = err_snake.gsub(/(?:^|_)([a-z])/) { $1.upcase }
  38     err_camel << 'Error' unless /Error\z/ =~ err_camel
  39     unless self.const_defined?(err_camel)
  40       self.class_eval("class #{err_camel} < MogileFS::Error; end")
  41     end
  42     BACKEND_ERRORS[err_snake] = self.const_get(err_camel)
  43   end
  44
  45   ##
  46   # The last error
  47
  48   attr_reader :lasterr
  49
  50   ##
  51   # The string attached to the last error
  52
  53   attr_reader :lasterrstr
  54
  55   ##
  56   # Creates a new MogileFS::Backend.
  57   #
  58   # :hosts is a required argument and must be an Array containing one or more
  59   # 'hostname:port' pairs as Strings.
  60   #
  61   # :timeout adjusts the request timeout before an error is returned.
  62
  63   def initialize(args)
  64     @hosts = args[:hosts]
  65     raise ArgumentError, "must specify at least one host" unless @hosts
  66     raise ArgumentError, "must specify at least one host" if @hosts.empty?
  67     unless @hosts == @hosts.select { |h| h =~ /:\d+$/ } then
  68       raise ArgumentError, ":hosts must be in 'host:port' form"
  69     end
  70
  71     @mutex = Mutex.new
  72     @timeout = args[:timeout] || 3
  73     @socket = nil
  74     @lasterr = nil
  75     @lasterrstr = nil
  76
  77     @dead = {}
  78   end
  79
  80   ##
  81   # Closes this backend's socket.
  82
  83   def shutdown
  84     @mutex.synchronize { shutdown_unlocked }
  85   end
  86
  87   # MogileFS::MogileFS commands
  88
  89   add_command :create_open
  90   add_command :create_close
  91   add_idempotent_command :get_paths
  92   add_command :delete
  93   add_idempotent_command :sleep
  94   add_command :rename
  95   add_idempotent_command :list_keys
  96   add_idempotent_command :file_info
  97   add_idempotent_command :file_debug
  98
  99   # MogileFS::Backend commands
 100
 101   add_idempotent_command :get_hosts
 102   add_idempotent_command :get_devices
 103   add_idempotent_command :list_fids
 104   add_idempotent_command :stats
 105   add_idempotent_command :get_domains
 106   add_command :create_domain
 107   add_command :delete_domain
 108   add_command :create_class
 109   add_command :update_class
 110   add_command :delete_class
 111   add_command :create_host
 112   add_command :update_host
 113   add_command :delete_host
 114   add_command :set_state
 115
 116   # Errors copied from MogileFS/Worker/Query.pm
 117   add_error 'dup'
 118   add_error 'after_mismatch'
 119   add_error 'bad_params'
 120   add_error 'class_exists'
 121   add_error 'class_has_files'
 122   add_error 'class_not_found'
 123   add_error 'db'
 124   add_error 'domain_has_files'
 125   add_error 'domain_exists'
 126   add_error 'domain_not_empty'
 127   add_error 'domain_not_found'
 128   add_error 'failure'
 129   add_error 'host_exists'
 130   add_error 'host_mismatch'
 131   add_error 'host_not_empty'
 132   add_error 'host_not_found'
 133   add_error 'invalid_chars'
 134   add_error 'invalid_checker_level'
 135   add_error 'invalid_mindevcount'
 136   add_error 'key_exists'
 137   add_error 'no_class'
 138   add_error 'no_devices'
 139   add_error 'no_domain'
 140   add_error 'no_host'
 141   add_error 'no_ip'
 142   add_error 'no_key'
 143   add_error 'no_port'
 144   add_error 'none_match'
 145   add_error 'plugin_aborted'
 146   add_error 'state_too_high'
 147   add_error 'size_verify_error'
 148   add_error 'unknown_command'
 149   add_error 'unknown_host'
 150   add_error 'unknown_key'
 151   add_error 'unknown_state'
 152   add_error 'unreg_domain'
 153
 154   private unless defined? $TESTING
 155
 156   def shutdown_unlocked # :nodoc:
 157     if @socket
 158       @socket.close rescue nil # ignore errors
 159       @socket = nil
 160     end
 161   end
 162
 163   ##
 164   # Performs the +cmd+ request with +args+.
 165
 166   def do_request(cmd, args, idempotent = false)
 167     request = make_request cmd, args
 168     @mutex.synchronize do
 169       begin
 170         io = socket
 171         begin
 172           io.timed_write(request, @timeout)
 173         rescue SystemCallError => err
 174           @dead[@active_host] = [ Time.now, err ]
 175           shutdown_unlocked
 176           io = socket
 177           retry
 178         end
 179
 180         line = io.timed_gets(@timeout) and return parse_response(line)
 181
 182         idempotent or
 183           raise EOFError, "end of file reached after: #{request.inspect}"
 184       rescue SystemCallError,
 185              MogileFS::UnreadableSocketError,
 186              MogileFS::InvalidResponseError, # truncated response
 187              MogileFS::Timeout
 188         # we got a successful timed_write, but not a timed_gets
 189         retry if idempotent
 190         shutdown_unlocked
 191         raise
 192       rescue => err
 193         # we DO NOT want the response we timed out waiting for, to crop up later
 194         # on, on the same socket, intersperesed with a subsequent request!  we
 195         # close the socket if there's any error.
 196         shutdown_unlocked
 197         raise
 198       end while idempotent
 199     end # @mutex.synchronize
 200   end
 201
 202   ##
 203   # Makes a new request string for +cmd+ and +args+.
 204
 205   def make_request(cmd, args)
 206     "#{cmd} #{url_encode args}\r\n"
 207   end
 208
 209   # this converts an error code from a mogilefsd tracker to an exception
 210   # Most of these exceptions should already be defined, but since the
 211   # MogileFS server code is liable to change and we may not always be
 212   # able to keep up with the changes
 213   def error(err_snake)
 214     BACKEND_ERRORS[err_snake] || self.class.add_error(err_snake)
 215   end
 216
 217   ##
 218   # Turns the +line+ response from the server into a Hash of options, an
 219   # error, or raises, as appropriate.
 220
 221   def parse_response(line)
 222     if line =~ /^ERR\s+(\w+)\s*([^\r\n]*)/
 223       @lasterr = $1
 224       @lasterrstr = $2 ? url_unescape($2) : nil
 225       raise error(@lasterr), @lasterrstr
 226     end
 227
 228     return url_decode($1) if line =~ /^OK\s+\d*\s*(\S*)\r\n\z/
 229
 230     raise MogileFS::InvalidResponseError,
 231           "Invalid response from server: #{line.inspect}"
 232   end
 233
 234   ##
 235   # Returns a socket connected to a MogileFS tracker.
 236
 237   def socket
 238     return @socket if @socket and not @socket.closed?
 239
 240     now = Time.now
 241
 242     @hosts.shuffle.each do |host|
 243       next if @dead.include?(host) and @dead[host][0] > now - 5
 244
 245       begin
 246         addr, port = host.split(/:/)
 247         @socket = MogileFS::Socket.tcp(addr, port, @timeout)
 248         @active_host = host
 249       rescue SystemCallError, MogileFS::Timeout => err
 250         @dead[host] = [ now, err ]
 251         next
 252       end
 253
 254       return @socket
 255     end
 256
 257     errors = @dead.map { |host,(_,e)| "#{host} - #{e.message} (#{e.class})" }
 258     raise MogileFS::UnreachableBackendError,
 259           "couldn't connect to any tracker: #{errors.join(', ')}"
 260   end
 261
 262   ##
 263   # Turns a url params string into a Hash.
 264
 265   def url_decode(str)
 266     Hash[*(str.split(/&/).map! { |pair|
 267       pair.split(/=/, 2).map! { |x| url_unescape(x) }
 268     } ).flatten]
 269   end
 270
 271   ##
 272   # Turns a Hash (or Array of pairs) into a url params string.
 273
 274   def url_encode(params)
 275     params.map do |k,v|
 276       "#{url_escape k.to_s}=#{url_escape v.to_s}"
 277     end.join("&")
 278   end
 279
 280   ##
 281   # Escapes naughty URL characters.
 282   if ''.respond_to?(:ord) # Ruby 1.9
 283     def url_escape(str)
 284       str.gsub(/([^\w\,\-.\/\\\: ])/) { "%%%02x" % $1.ord }.tr(' ', '+')
 285     end
 286   else # Ruby 1.8
 287     def url_escape(str)
 288       str.gsub(/([^\w\,\-.\/\\\: ])/) { "%%%02x" % $1[0] }.tr(' ', '+')
 289     end
 290   end
 291
 292   ##
 293   # Unescapes naughty URL characters.
 294
 295   def url_unescape(str)
 296     str.gsub(/%([a-f0-9][a-f0-9])/i) { [$1.to_i(16)].pack 'C' }.tr('+', ' ')
 297   end
 298 end