Check that a mirror write returning an error will cause a reconnect and retry

src/readwrite.c

@@ -68,16 +68,17 @@ void fill_request(struct nbd_request *request, int type, int from, int len)
 
 void read_reply(int fd, struct nbd_request *request, struct nbd_reply *reply)
 {
-	FATAL_IF_NEGATIVE(readloop(fd, reply, sizeof(*reply)),
+	ERROR_IF_NEGATIVE(readloop(fd, reply, sizeof(*reply)),
 	  "Couldn't read reply");
+
 	if (be32toh(reply->magic) != REPLY_MAGIC) {
-		fatal("Reply magic incorrect (%p)", be32toh(reply->magic));
+		error("Reply magic incorrect (%p)", be32toh(reply->magic));
 	}
 	if (be32toh(reply->error) != 0) {
-		fatal("Server replied with error %d", be32toh(reply->error));
+		error("Server replied with error %d", be32toh(reply->error));
 	}
 	if (strncmp(request->handle, reply->handle, 8) != 0) {
-		fatal("Did not reply with correct handle");
+		error("Did not reply with correct handle");
 	}
 }
 

tests/fakes/dest/close_after_hello.rb

@@ -7,19 +7,16 @@
 # user, we have to keep trying.
 
 require 'flexnbd/fake_dest'
-include FlexNBD::FakeDest
+include FlexNBD
 
-addr, port = *ARGV
+server = FakeDest.new( *ARGV )
+client = server.accept( "Timed out waiting for a connection" )
+client.write_hello
+client.close
 
+new_client = server.accept( "Timed out waiting for a reconnection" )
+new_client.close
 
-sock = serve( addr, port )
+server.close
-client_sock = accept( sock, "Timed out waiting for a connection" )
-write_hello( client_sock )
-client_sock.close
-
-new_sock = accept( sock, "Timed out waiting for a reconnection" )
-
-new_sock.close
-sock.close
 
 exit 0

tests/fakes/dest/error_on_write.rb

@@ -0,0 +1,21 @@
+#!/usr/bin/env ruby
+# encoding: utf-8
+
+require 'flexnbd/fake_dest'
+include FlexNBD
+
+server = FakeDest.new( *ARGV )
+client = server.accept
+
+client.write_hello
+handle = client.read_request[:handle]
+client.write_error( handle )
+
+
+client2 = server.accept( "Timed out waiting for a reconnection" )
+
+client.close
+client2.close
+server.close
+
+exit(0)

tests/fakes/dest/hang_after_connect.rb

@@ -8,16 +8,14 @@
 # right error message after the timeout time.
 
 require 'flexnbd/fake_dest'
-include FlexNBD::FakeDest
+include FlexNBD
 
-addr, port = *ARGV
+server = FakeDest.new( *ARGV )
-
+client = server.accept( "Client didn't make a connection" )
-serve_sock = serve( addr, port )
-client_sock = accept( serve_sock, "Client didn't make a connection" )
 
 # Sleep for one second past the timeout (a bit of slop in case ruby
 # doesn't launch things quickly)
 sleep(FlexNBD::MS_HELLO_TIME_SECS + 1)
 
-client_sock.close  if client_sock
+client.close
-serve_sock.close
+server.close

tests/fakes/dest/hang_after_write.rb

@@ -6,22 +6,23 @@
 # write has gone MIA, and we expect a reconnect.
 
 require 'flexnbd/fake_dest'
-include FlexNBD::FakeDest
+include FlexNBD
 
-sock = serve( *ARGV )
+server = FakeDest.new( *ARGV )
-client_sock1 = accept( sock )
+client1 = server.accept( server )
-write_hello( client_sock1 )
+client1.write_hello
-read_request( client_sock1 )
+client1.read_request
 
 t = Thread.start do
-  client_sock2 = accept( sock, "Timed out waiting for a reconnection",
+  client2 = server.accept( "Timed out waiting for a reconnection",
                         FlexNBD::MS_REQUEST_LIMIT_SECS + 2 )
-  client_sock2.close
+  client2.close
 end
 
 sleep( FlexNBD::MS_REQUEST_LIMIT_SECS + 2 )
-client_sock1.close
+client1.close
 
 t.join
 
+server.close
 exit(0)

tests/fakes/dest/hello_wrong_magic.rb

@@ -4,21 +4,21 @@
 # We expect the sender to disconnect and reconnect.
 
 require 'flexnbd/fake_dest'
-include FlexNBD::FakeDest
+include FlexNBD
 
-sock = serve( *ARGV )
+server = FakeDest.new( *ARGV )
-client_sock = accept( sock, "Timed out waiting for a connection" )
+client1 = server.accept
 
 # Launch a second thread so that we can spot the reconnection attempt
 # as soon as it happens, or alternatively die a flaming death on
 # timeout.
 t = Thread.new do
-  client_sock2 = accept( sock, "Timed out waiting for a reconnection",
+  client2 = server.accept( "Timed out waiting for a reconnection",
-                         FlexNBD::MS_RETRY_DELAY_SECS + 1 )
+                           FlexNBD::MS_RETRY_DELAY_SECS + 1 )
-  client_sock2.close
+  client2.close
 end
 
-write_hello( client_sock, :magic => :wrong )
+client1.write_hello( :magic => :wrong )
 
 t.join
 

tests/fakes/dest/hello_wrong_size.rb

@@ -5,19 +5,18 @@
 # EOF on read.
 
 require 'flexnbd/fake_dest'
-include FlexNBD::FakeDest
+include FlexNBD
-
-sock = serve( *ARGV )
-client_sock = accept( sock, "Timed out waiting for a connection" )
 
+server = FakeDest.new( *ARGV )
+client = server.accept
 
 t = Thread.new do
-  client_sock2 = accept( sock, "Timed out waiting for a reconnection",
+  client2 = server.accept( "Timed out waiting for a reconnection",
-                         FlexNBD::MS_RETRY_DELAY_SECS + 1 )
+                    FlexNBD::MS_RETRY_DELAY_SECS + 1 )
-  client_sock2.close
+  client2.close
 end
 
-write_hello( client_sock, :size => :wrong )
+client.write_hello( :size => :wrong )
 
 t.join
 

tests/fakes/dest/reject_acl.rb

@@ -3,11 +3,11 @@
 # Accept a connection, then immediately close it.  This simulates an ACL rejection.
 
 require 'flexnbd/fake_dest'
-include FlexNBD::FakeDest
+include FlexNBD
 
-serve_sock = serve( *ARGV )
+server = FakeDest.new( *ARGV )
-accept( serve_sock, "Timed out waiting for a connection" ).close
+server.accept.close
 
-serve_sock.close
+server.close
 
 exit(0)

tests/flexnbd/fake_dest.rb

@@ -6,14 +6,108 @@ require 'timeout'
 require 'flexnbd/constants'
 
 module FlexNBD
-  module FakeDest
+  class FakeDest
 
-    def serve( addr, port )
+    class Client
+      def initialize( sock )
+        @sock = sock
+      end
+
+
+      def write_hello( opts = {} )
+        self.class.write_hello( @sock, opts )
+      end
+
+      def read_request()
+        self.class.read_request( @sock )
+      end
+
+      def write_error( handle )
+        self.class.write_error( @sock, handle )
+      end
+
+      def close
+        @sock.close
+      end
+
+
+      def self.write_hello( client_sock, opts={} )
+        client_sock.write( "NBDMAGIC" )
+
+        if opts[:magic] == :wrong
+          client_sock.write( "\x00\x00\x42\x02\x81\x86\x12\x52" )
+        else
+          client_sock.write( "\x00\x00\x42\x02\x81\x86\x12\x53" )
+        end
+
+        if opts[:size] == :wrong
+          8.times do client_sock.write rand(256).chr end
+        else
+          client_sock.write( "\x00\x00\x00\x00\x00\x00\x10\x00" )
+        end
+
+        client_sock.write( "\x00" * 128 )
+      end
+
+
+
+      def self.read_request( client_sock )
+        req = client_sock.read(28)
+
+        magic_s  = req[0  ... 4 ]
+        type_s   = req[4  ... 8 ]
+        handle_s = req[8  ... 16]
+        from_s   = req[16 ... 24]
+        len_s    = req[24 ... 28]
+
+        {
+          :magic  => magic_s,
+          :type   => type_s.unpack("N").first,
+          :handle => handle_s,
+          :from   => parse_be64( from_s ),
+          :len    => len_s.unpack( "N").first
+        }
+      end
+
+
+      def self.parse_be64(str)
+        raise "String is the wrong length: 8 bytes expected (#{str.length} received)" unless
+          str.length == 8
+
+        top, bottom = str.unpack("NN")
+        (top << 32) + bottom
+      end
+
+
+      def self.write_error( client_sock, handle )
+        client_sock.write( "\x67\x44\x66\x98")
+        client_sock.write( "\x00\x00\x00\x01")
+        client_sock.write( handle )
+      end
+
+
+    end # class Client
+
+
+    def initialize( addr, port )
+      @sock = self.class.serve( addr, port )
+    end
+
+    def accept( err_msg = "Timed out waiting for a connection", timeout = 2)
+      Client.new( self.class.accept( @sock, err_msg, timeout ) )
+    end
+
+    def close
+      @sock.close
+    end
+
+
+    def self.serve( addr, port )
       TCPServer.new( addr, port )
     end
 
 
-    def accept( sock, err_msg = "Timed out waiting for a connection", timeout=2 )
+    def self.accept( sock, err_msg, timeout )
       client_sock = nil
 
       begin
@@ -29,29 +123,5 @@ module FlexNBD
     end
 
 
-    def write_hello( client_sock, opts={} )
-      client_sock.write( "NBDMAGIC" )
-
-      if opts[:magic] == :wrong
-        client_sock.write( "\x00\x00\x42\x02\x81\x86\x12\x52" )
-      else
-        client_sock.write( "\x00\x00\x42\x02\x81\x86\x12\x53" )
-      end
-
-      if opts[:size] == :wrong
-        8.times do client_sock.write rand(256).chr end
-      else
-        client_sock.write( "\x00\x00\x00\x00\x00\x00\x10\x00" )
-      end
-
-      client_sock.write( "\x00" * 128 )
-    end
-
-
-    def read_request( client_sock )
-      client_sock.read(28)
-    end
-
-
   end # module FakeDest
 end # module FlexNBD

tests/nbd_scenarios

@@ -233,52 +233,69 @@ class NBDConnectSourceFailureScenarios < Test::Unit::TestCase
 
 
   def test_destination_hangs_after_connect_reports_error_at_source
-    @env.run_fake( "dest/hang_after_connect", @env.ip, @env.port2 )
+    run_fake( "dest/hang_after_connect" )
 
     stdout, stderr = @env.mirror12_unchecked
     assert_match( /Remote server failed to respond/, stderr )
-    assert @env.fake_reports_success
+    assert_success
   end
 
 
   def test_destination_rejects_connection_reports_error_at_source
-    @env.run_fake( "dest/reject_acl", @env.ip, @env.port2 )
+    run_fake( "dest/reject_acl" )
 
     stdout, stderr = @env.mirror12_unchecked
     assert_match /Mirror was rejected/, stderr
-    assert @env.fake_reports_success
+    assert_success
   end
 
   def test_wrong_size_causes_disconnect
-    @env.run_fake( "dest/hello_wrong_size", @env.ip, @env.port2 )
+    run_fake( "dest/hello_wrong_size" )
     stdout, stderr = @env.mirror12_unchecked
     assert_match /Remote size does not match local size/, stderr
-    assert @env.fake_reports_success
+    assert_success
   end
 
 
   def test_wrong_magic_causes_disconnect
-    @env.run_fake( "dest/hello_wrong_magic", @env.ip, @env.port2 )
+    run_fake( "dest/hello_wrong_magic" )
     stdout, stderr = @env.mirror12_unchecked
     assert_match /Mirror was rejected/, stderr
-    assert @env.fake_reports_success, "dest/hello_wrong_magic fake failed"
+    assert_success "dest/hello_wrong_magic fake failed"
   end
 
 
   def test_disconnect_after_hello_causes_retry
-    @env.run_fake( "dest/close_after_hello", @env.ip, @env.port2 )
+    run_fake( "dest/close_after_hello" )
     stdout, stderr = @env.mirror12_unchecked
     assert_match( /Mirror started/, stdout )
 
-    assert @env.fake_reports_success
+    assert_success
   end
 
 
   def test_write_times_out_causes_retry
-    @env.run_fake( "dest/hang_after_write", @env.ip, @env.port2 )
+    run_fake( "dest/hang_after_write" )
     stdout, stderr = @env.mirror12_unchecked
 
-    assert @env.fake_reports_success, "Fake failed."
+    assert_success
+  end
+
+
+  def test_rejected_write_causes_retry
+    run_fake( "dest/error_on_write" )
+    stdout, stderr = @env.mirror12_unchecked
+    assert_success
+  end
+
+
+  private
+  def run_fake(name)
+    @env.run_fake( name, @env.ip, @env.port2 )
+  end
+
+  def assert_success( msg=nil )
+    assert @env.fake_reports_success, msg || "Fake failed"
   end