lib/mogilefs/backend.rb

   1 # -*- encoding: binary -*-
   2 require 'thread'
   3
   4 # This class communicates with the MogileFS trackers.
   5 # You should not have to use this directly unless you are developing
   6 # support for new commands or plugins for MogileFS
   7 class MogileFS::Backend
   8
   9   # Adds MogileFS commands +names+.
  10   def self.add_command(*names)
  11     names.each do |name|
  12       define_method name do |*args|
  13         do_request(name, args[0] || {}, false)
  14       end
  15     end
  16   end
  17
  18   # adds idempotent MogileFS commands +names+, these commands may be retried
  19   # transparently on a different tracker if there is a network/server error.
  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 = {} # :nodoc:
  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 const_defined?(err_camel)
  40       const_set(err_camel, Class.new(MogileFS::Error))
  41     end
  42     BACKEND_ERRORS[err_snake] = const_get(err_camel)
  43   end
  44
  45   def self.const_missing(name) # :nodoc:
  46     if /Error\z/ =~ name.to_s
  47       const_set(name, Class.new(MogileFS::Error))
  48     else
  49       super name
  50     end
  51   end
  52
  53   ##
  54   # The last error
  55
  56   attr_reader :lasterr
  57
  58   ##
  59   # The string attached to the last error
  60
  61   attr_reader :lasterrstr
  62
  63   ##
  64   # Creates a new MogileFS::Backend.
  65   #
  66   # :hosts is a required argument and must be an Array containing one or more
  67   # 'hostname:port' pairs as Strings.
  68   #
  69   # :timeout adjusts the request timeout before an error is returned.
  70
  71   def initialize(args)
  72     @hosts = args[:hosts]
  73     @fail_timeout = args[:fail_timeout] || 5
  74     raise ArgumentError, "must specify at least one host" unless @hosts
  75     raise ArgumentError, "must specify at least one host" if @hosts.empty?
  76     unless @hosts == @hosts.select { |h| h =~ /:\d+$/ } then
  77       raise ArgumentError, ":hosts must be in 'host:port' form"
  78     end
  79
  80     @mutex = Mutex.new
  81     @timeout = args[:timeout] || 3
  82     @socket = nil
  83     @lasterr = nil
  84     @lasterrstr = nil
  85     @pending = []
  86
  87     @dead = {}
  88   end
  89
  90   ##
  91   # Closes this backend's socket.
  92
  93   def shutdown
  94     @mutex.synchronize { shutdown_unlocked }
  95   end
  96
  97   # MogileFS::MogileFS commands
  98
  99   add_command :create_open
 100   add_command :create_close
 101   add_idempotent_command :get_paths
 102   add_idempotent_command :noop
 103   add_command :delete
 104   add_idempotent_command :sleep
 105   add_command :rename
 106   add_idempotent_command :list_keys
 107   add_idempotent_command :file_info
 108   add_idempotent_command :file_debug
 109
 110   # MogileFS::Backend commands
 111
 112   add_idempotent_command :get_hosts
 113   add_idempotent_command :get_devices
 114   add_idempotent_command :list_fids
 115   add_idempotent_command :stats
 116   add_idempotent_command :get_domains
 117   add_command :create_device
 118   add_command :create_domain
 119   add_command :delete_domain
 120   add_command :create_class
 121   add_command :update_class
 122   add_command :updateclass
 123   add_command :delete_class
 124   add_command :create_host
 125   add_command :update_host
 126   add_command :delete_host
 127   add_command :set_state
 128   add_command :set_weight
 129   add_command :replicate_now
 130
 131   def shutdown_unlocked(do_raise = false) # :nodoc:
 132     @pending = []
 133     if @socket
 134       @socket.close rescue nil # ignore errors
 135       @socket = nil
 136     end
 137     raise if do_raise
 138   end
 139
 140   def dispatch_unlocked(request, timeout = @timeout) # :nodoc:
 141     begin
 142       io = socket
 143       io.timed_write(request, timeout)
 144       io
 145     rescue SystemCallError, MogileFS::RequestTruncatedError  => err
 146       @dead[@active_host] = [ Time.now, err ]
 147       shutdown_unlocked
 148       retry
 149     end
 150   end
 151
 152   def pipeline_gets_unlocked(io, timeout) # :nodoc:
 153     line = io.timed_gets(timeout) or
 154       raise MogileFS::PipelineError,
 155             "EOF with #{@pending.size} requests in-flight"
 156     ready = @pending.shift
 157     ready[1].call(parse_response(line, ready[0]))
 158   end
 159
 160   def timeout_update(timeout, t0) # :nodoc:
 161     timeout -= (Time.now - t0)
 162     timeout < 0 ? 0 : timeout
 163   end
 164
 165   # try to read any responses we have pending already before filling
 166   # the pipeline more requests.  This usually takes very little time,
 167   # but trackers may return huge responses and we could be on a slow
 168   # network.
 169   def pipeline_drain_unlocked(io, timeout) # :nodoc:
 170     set = [ io ]
 171     while @pending.size > 0
 172       t0 = Time.now
 173       r = IO.select(set, set, nil, timeout)
 174       timeout = timeout_update(timeout, t0)
 175
 176       if r && r[0][0]
 177         t0 = Time.now
 178         pipeline_gets_unlocked(io, timeout)
 179         timeout = timeout_update(timeout, t0)
 180       else
 181         return timeout
 182       end
 183     end
 184     timeout
 185   end
 186
 187   # dispatch a request like do_request, but queue +block+ for execution
 188   # upon receiving a response.  It is the users' responsibility to ensure
 189   # &block is executed in the correct order.  Trackers with multiple
 190   # queryworkers are not guaranteed to return responses in the same
 191   # order they were requested.
 192   def pipeline_dispatch(cmd, args, &block) # :nodoc:
 193     request = make_request(cmd, args)
 194     timeout = @timeout
 195
 196     @mutex.synchronize do
 197       io = socket
 198       timeout = pipeline_drain_unlocked(io, timeout)
 199
 200       # send the request out...
 201       begin
 202         io.timed_write(request, timeout)
 203         @pending << [ request, block ]
 204       rescue SystemCallError, MogileFS::RequestTruncatedError => err
 205         @dead[@active_host] = [ Time.now, err ]
 206         shutdown_unlocked(@pending[0])
 207         io = socket
 208         retry
 209       end
 210
 211       @pending.size
 212     end
 213   end
 214
 215   def pipeline_wait(count = nil) # :nodoc:
 216     @mutex.synchronize do
 217       io = socket
 218       count ||= @pending.size
 219       @pending.size < count and
 220         raise MogileFS::Error,
 221               "pending=#{@pending.size} < expected=#{count} failed"
 222       begin
 223         count.times { pipeline_gets_unlocked(io, @timeout) }
 224       rescue
 225         shutdown_unlocked(true)
 226       end
 227     end
 228   end
 229
 230   # Performs the +cmd+ request with +args+.
 231   def do_request(cmd, args, idempotent = false)
 232     no_raise = args.delete(:ruby_no_raise)
 233     request = make_request(cmd, args)
 234     line = nil
 235     failed = false
 236     @mutex.synchronize do
 237       begin
 238         io = dispatch_unlocked(request)
 239         line = io.timed_gets(@timeout)
 240         break if /\r?\n\z/ =~ line
 241
 242         line and raise MogileFS::InvalidResponseError,
 243                        "Invalid response from server: #{line.inspect}"
 244
 245         idempotent or
 246           raise EOFError, "end of file reached after: #{request.inspect}"
 247         # fall through to retry in loop
 248       rescue SystemCallError,
 249              MogileFS::UnreadableSocketError,
 250              MogileFS::InvalidResponseError, # truncated response
 251              MogileFS::Timeout
 252         # we got a successful timed_write, but not a timed_gets
 253         if idempotent
 254           failed = true
 255           retry
 256         end
 257         shutdown_unlocked(true)
 258       rescue
 259         # we DO NOT want the response we timed out waiting for, to crop up later
 260         # on, on the same socket, intersperesed with a subsequent request!  we
 261         # close the socket if there's any error.
 262         shutdown_unlocked(true)
 263       end while idempotent
 264       shutdown_unlocked if failed
 265     end # @mutex.synchronize
 266     parse_response(line, no_raise ? request : nil)
 267   end
 268
 269   # Makes a new request string for +cmd+ and +args+.
 270   def make_request(cmd, args)
 271     "#{cmd} #{url_encode args}\r\n"
 272   end
 273
 274   # this converts an error code from a mogilefsd tracker to an exception
 275   # Most of these exceptions should already be defined, but since the
 276   # MogileFS server code is liable to change and we may not always be
 277   # able to keep up with the changes
 278   def error(err_snake)
 279     BACKEND_ERRORS[err_snake] || self.class.add_error(err_snake)
 280   end
 281
 282   # Turns the +line+ response from the server into a Hash of options, an
 283   # error, or raises, as appropriate.
 284   def parse_response(line, request = nil)
 285     case line
 286     when /\AOK\s+\d*\s*(\S*)\r?\n\z/
 287       url_decode($1)
 288     when /\AERR\s+(\w+)\s*([^\r\n]*)/
 289       @lasterr = $1
 290       @lasterrstr = $2 ? url_unescape($2) : nil
 291       if request
 292         request = " request=#{request.strip}"
 293         @lasterrstr = @lasterrstr ? (@lasterrstr << request) : request
 294         return error(@lasterr).new(@lasterrstr)
 295       end
 296       raise error(@lasterr).new(@lasterrstr)
 297     else
 298       raise MogileFS::InvalidResponseError,
 299             "Invalid response from server: #{line.inspect}"
 300     end
 301   end
 302
 303   # this command is special since the cache is per-tracker, so we connect
 304   # to all backends and not just one
 305   def clear_cache(types = %w(all))
 306     opts = {}
 307     types.each { |type| opts[type] = 1 }
 308
 309     sockets = @hosts.map do |host|
 310       MogileFS::Socket.start(*(host.split(/:/))) rescue nil
 311     end
 312     sockets.compact!
 313
 314     wpending = sockets
 315     rpending = []
 316     request = make_request("clear_cache", opts)
 317     while wpending[0] || rpending[0]
 318       r = IO.select(rpending, wpending, nil, @timeout) or return
 319       rpending -= r[0]
 320       wpending -= r[1]
 321       r[0].each { |io| io.timed_gets(0) rescue nil }
 322       r[1].each do |io|
 323         begin
 324           io.timed_write(request, 0)
 325           rpending << io
 326         rescue
 327         end
 328       end
 329     end
 330     nil
 331     ensure
 332       sockets.each { |io| io.close }
 333   end
 334
 335   # Returns a socket connected to a MogileFS tracker.
 336   def socket
 337     return @socket if @socket and not @socket.closed?
 338
 339     @hosts.shuffle.each do |host|
 340       next if dead = @dead[host] and dead[0] > (Time.now - @fail_timeout)
 341
 342       begin
 343         addr, port = host.split(/:/)
 344         @socket = MogileFS::Socket.tcp(addr, port, @timeout)
 345         @active_host = host
 346       rescue SystemCallError, MogileFS::Timeout => err
 347         @dead[host] = [ Time.now, err ]
 348         next
 349       end
 350
 351       return @socket
 352     end
 353
 354     errors = @dead.map { |host,(_,e)| "#{host} - #{e.message} (#{e.class})" }
 355     raise MogileFS::UnreachableBackendError,
 356           "couldn't connect to any tracker: #{errors.join(', ')}"
 357   end
 358
 359   # Turns a url params string into a Hash.
 360   def url_decode(str) # :nodoc:
 361     rv = {}
 362     str.split(/&/).each do |pair|
 363       k, v = pair.split(/=/, 2).map! { |x| url_unescape(x) }
 364       rv[k.freeze] = v
 365     end
 366     rv
 367   end
 368
 369   # :stopdoc:
 370   # TODO: see if we can use existing URL-escape/unescaping routines
 371   # in the Ruby standard library, Perl MogileFS seems to NIH these
 372   #  routines, too
 373   # :startdoc:
 374
 375   # Turns a Hash (or Array of pairs) into a url params string.
 376   def url_encode(params) # :nodoc:
 377     params.map do |k,v|
 378       "#{url_escape k.to_s}=#{url_escape v.to_s}"
 379     end.join("&")
 380   end
 381
 382   # Escapes naughty URL characters.
 383   if ''.respond_to?(:ord) # Ruby 1.9
 384     def url_escape(str) # :nodoc:
 385       str.gsub(/([^\w\,\-.\/\\\: ])/) { "%%%02x" % $1.ord }.tr(' ', '+')
 386     end
 387   else # Ruby 1.8
 388     def url_escape(str) # :nodoc:
 389       str.gsub(/([^\w\,\-.\/\\\: ])/) { "%%%02x" % $1[0] }.tr(' ', '+')
 390     end
 391   end
 392
 393   # Unescapes naughty URL characters.
 394   def url_unescape(str) # :nodoc:
 395     str.tr('+', ' ').gsub(/%([a-f0-9][a-f0-9])/i) { [$1.to_i(16)].pack 'C' }
 396   end
 397 end