lib/mogilefs/network.rb

   1 require 'mogilefs'
   2 require 'socket'
   3 require 'mogilefs/util'
   4
   5 module MogileFS::Network
   6   # given an array of URIs, verify that at least one of them is accessible
   7   # with the expected HTTP code within the timeout period (in seconds).
   8   def verify_uris(uris = [], expect = '200', timeout = 2.00)
   9     uri_socks = {}
  10
  11     # first, we asynchronously connect to all of them
  12     uris.each do |uri|
  13       sock = Socket.mogilefs_new_nonblock(uri.host, uri.port) rescue next
  14       uri_socks[sock] = uri
  15     end
  16
  17     # wait for at least one of them to finish connecting and send
  18     # HTTP requests to the connected ones
  19     sockets, timeout = get_writable_set(uri_socks, timeout)
  20
  21     # Await a response from the sockets we had written to, we only need one
  22     # valid response, but we'll take more if they return simultaneously
  23     sockets[0] ? get_readable_uris(sockets, uri_socks, expect, timeout) : []
  24
  25     ensure
  26       uri_socks.keys.each { |sock| sock.close rescue nil }
  27   end
  28
  29   private
  30     include MogileFS::Util
  31
  32     # returns an array of writeable Sockets and leftover from timeout
  33     def get_writable_set(uri_socks, timeout)
  34       sockets = []
  35       begin
  36         t0 = Time.now
  37         r = begin
  38          IO.select(nil, uri_socks.keys, nil, timeout > 0 ? timeout : 0)
  39         rescue
  40           # get rid of bad descriptors
  41           uri_socks.delete_if do |sock, uri|
  42             begin
  43               sock.recv_nonblock(1)
  44               false # should never get here for HTTP, really...
  45             rescue Errno::EAGAIN, Errno::EINTR
  46               false
  47             rescue
  48               sock.close rescue nil
  49               true
  50             end
  51           end
  52           timeout -= (Time.now - t0)
  53           retry if timeout >= 0
  54         end
  55
  56         break unless r && r[1]
  57
  58         r[1].each do |sock|
  59           begin
  60             # we don't care about short/interrupted writes here, if the
  61             # following request fails or blocks then the server is
  62             # flat-out hopeless
  63             sock.write_nonblock(
  64               "HEAD #{uri_socks[sock].request_uri} HTTP/1.0\r\n\r\n")
  65             sockets << sock
  66           rescue
  67             sock.close rescue nil
  68           end
  69         end
  70
  71         timeout -= (Time.now - t0)
  72       end until (sockets[0] || timeout < 0)
  73
  74       [ sockets, timeout ]
  75     end
  76
  77     # returns an array of URIs from uri_socks that are good
  78     def get_readable_uris(sockets, uri_socks, expect, timeout)
  79       ok_uris = []
  80
  81       begin
  82         t0 = Time.now
  83         r = IO.select(sockets, nil, nil, timeout > 0 ? timeout : 0) rescue nil
  84
  85         (r && r[0] ? r[0] : sockets).each do |sock|
  86           buf = begin
  87             sock.recv_nonblock(128, Socket::MSG_PEEK)
  88           rescue Errno::EAGAIN, Errno::EINTR
  89             next
  90           rescue
  91             sockets.delete(sock) # socket went bad
  92             next
  93           end
  94
  95           if buf && /\AHTTP\/[\d\.]+ #{expect} / =~ buf
  96             ok_uris << uri_socks.delete(sock)
  97             sock.close rescue nil
  98           end
  99         end
 100         timeout -= (Time.now - t0)
 101       end until ok_uris[0] || timeout < 0
 102
 103       ok_uris
 104     end
 105
 106 end # module MogileFS::Network