lib/mogilefs/network.rb

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