backend: respect timeout on socket/timeout errors

While retrying idempotent requests (even on timeouts) would
prevent stale sockets from being noticed, it is better to kill
the socket and immediately propagate the timeout error to the
user.  Retrying in a timeout may cause a request/response to
take longer (perhaps _much_ longer) than the timeout configured
by the user.

2 files changed, 35 insertions, 16 deletions

diff --git a/lib/mogilefs/backend.rb b/lib/mogilefs/backend.rb
index cbeec3d..afc2f1a 100644
--- a/lib/mogilefs/backend.rb
+++ b/lib/mogilefs/backend.rb
@@ -246,15 +246,16 @@ class MogileFS::Backend
           raise EOFError, "end of file reached after: #{request.inspect}"
         # fall through to retry in loop
       rescue SystemCallError,
-             MogileFS::UnreadableSocketError,
-             MogileFS::InvalidResponseError, # truncated response
-             MogileFS::Timeout
+             MogileFS::InvalidResponseError # truncated response
         # we got a successful timed_write, but not a timed_gets
         if idempotent
           failed = true
+          shutdown_unlocked(false)
           retry
         end
         shutdown_unlocked(true)
+      rescue MogileFS::UnreadableSocketError, MogileFS::Timeout
+        shutdown_unlocked(true)
       rescue
         # we DO NOT want the response we timed out waiting for, to crop up later
         # on, on the same socket, intersperesed with a subsequent request!  we
diff --git a/test/test_mogilefs.rb b/test/test_mogilefs.rb
index 249c2d6..439a045 100644
--- a/test/test_mogilefs.rb
+++ b/test/test_mogilefs.rb
@@ -609,30 +609,48 @@ class TestMogileFS__MogileFS < TestMogileFS
     ip = "127.0.0.1"
     a = TCPServer.new(ip, 0)
     hosts = [ "#{ip}:#{a.addr[1]}" ]
+    q = Queue.new
     timeout = 1
     args = { :hosts => hosts, :domain => "foo", :timeout => timeout }
     c = MogileFS::MogileFS.new(args)
     received = []
     secs = timeout + 1
     th = Thread.new do
-      loop {
-        x = a.accept
-        2.times do |i|
-          line = x.gets
-          %r{key=(\w+)} =~ line
-          sleep secs
-          secs = 0
-          x.write("OK paths=1&path1=http://0/#{$1}\r\n")
-        end
+      x = a.accept
+      line = x.gets
+      %r{key=(\w+)} =~ line
+      sleep(secs)
+      begin
+        x.write("OK paths=1&path1=http://0/#{$1}\r\n")
+      rescue Errno::EPIPE
+        # EPIPE may or may not get raised due to timing issue,
+        # we don't care either way
+      rescue => e
+        flunk("#{e.message} (#{e.class})")
+      ensure
+        x.close
+      end
+      q << :continue_test
+
+      x = a.accept
+      line = x.gets
+      %r{key=(\w+)} =~ line
+      begin
+        x.write("OK paths=1&path1=http://0/#{$1}\r\n")
+      rescue => e
+        flunk("#{e.message} (#{e.class})")
+      ensure
         x.close
-      }
+      end
+    end
+    assert_raises(MogileFS::UnreadableSocketError) do
+      c.get_paths("a")
     end
-    expect1 = %w(http://0/a)
-    assert_equal expect1, c.get_paths("a")
+    assert_equal :continue_test, q.pop, "avoid race during test"
     expect2 = %w(http://0/b)
     assert_equal expect2, c.get_paths("b")
     a.close
-    th.kill
+    th.join
   end
 
   def test_idempotent_command_response_truncated