api: graceful Connection.Close()

Connection.Close() waits for all active requests to complete.

Part of #257

Author: oleg-jukovec

CHANGELOG.md

@@ -12,6 +12,8 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 
 ### Changed
 
+- Connection.Close() waits for all active requests to complete (#257)
+
 ### Fixed
 
 - crud tests with Tarantool 3.0 (#293)

connection.go

@@ -37,10 +37,10 @@ const (
 	Disconnected
 	// ReconnectFailed signals that attempt to reconnect has failed.
 	ReconnectFailed
-	// Either reconnect attempts exhausted, or explicit Close is called.
-	Closed
 	// Shutdown signals that shutdown callback is processing.
 	Shutdown
+	// Either reconnect attempts exhausted, or explicit Close is called.
+	Closed
 
 	// LogReconnectFailed is logged when reconnect attempt failed.
 	LogReconnectFailed ConnLogKind = iota + 1
@@ -456,10 +456,7 @@ func (conn *Connection) ClosedNow() bool {
 // Close closes Connection.
 // After this method called, there is no way to reopen this Connection.
 func (conn *Connection) Close() error {
-	err := ClientError{ErrConnectionClosed, "connection closed by client"}
-	conn.mutex.Lock()
-	defer conn.mutex.Unlock()
-	return conn.closeConnection(err, true)
+	return conn.shutdown(true)
 }
 
 // Addr returns a configured address of Tarantool socket.
@@ -1532,17 +1529,27 @@ func shutdownEventCallback(event WatchEvent) {
 	// step 2.
 	val, ok := event.Value.(bool)
 	if ok && val {
-		go event.Conn.shutdown()
+		go event.Conn.shutdown(false)
 	}
 }
 
-func (conn *Connection) shutdown() {
+func (conn *Connection) shutdown(forever bool) error {
 	// Forbid state changes.
 	conn.mutex.Lock()
 	defer conn.mutex.Unlock()
 
 	if !atomic.CompareAndSwapUint32(&conn.state, connConnected, connShutdown) {
-		return
+		if forever {
+			err := ClientError{ErrConnectionClosed, "connection closed by client"}
+			return conn.closeConnection(err, true)
+		}
+		return nil
+	}
+
+	if forever {
+		// We don't want to reconnect any more.
+		conn.opts.Reconnect = 0
+		conn.opts.MaxReconnects = 0
 	}
 
 	conn.cond.Broadcast()
@@ -1551,7 +1558,7 @@ func (conn *Connection) shutdown() {
 	c := conn.c
 	for {
 		if (atomic.LoadUint32(&conn.state) != connShutdown) || (c != conn.c) {
-			return
+			return nil
 		}
 		if atomic.LoadInt64(&conn.requestCnt) == 0 {
 			break
@@ -1563,14 +1570,19 @@ func (conn *Connection) shutdown() {
 		conn.cond.Wait()
 	}
 
-	// Start to reconnect based on common rules, same as in net.box.
-	// Reconnect also closes the connection: server waits until all
-	// subscribed connections are terminated.
-	// See https://www.tarantool.io/en/doc/latest/dev_guide/internals/iproto/graceful_shutdown/
-	// step 3.
-	conn.reconnectImpl(
-		ClientError{
+	if forever {
+		err := ClientError{ErrConnectionClosed, "connection closed by client"}
+		return conn.closeConnection(err, true)
+	} else {
+		// Start to reconnect based on common rules, same as in net.box.
+		// Reconnect also closes the connection: server waits until all
+		// subscribed connections are terminated.
+		// See https://www.tarantool.io/en/doc/latest/dev_guide/internals/iproto/graceful_shutdown/
+		// step 3.
+		conn.reconnectImpl(ClientError{
 			ErrConnectionClosed,
 			"connection closed after server shutdown",
 		}, conn.c)
+		return nil
+	}
 }

shutdown_test.go

@@ -143,6 +143,42 @@ func TestGracefulShutdown(t *testing.T) {
 	testGracefulShutdown(t, conn, &inst)
 }
 
+func TestGracefulShutdownOnClose(t *testing.T) {
+	test_helpers.SkipIfWatchersUnsupported(t)
+
+	var inst test_helpers.TarantoolInstance
+	var conn *Connection
+	var err error
+
+	inst, err = test_helpers.StartTarantool(shtdnSrvOpts)
+	require.Nil(t, err)
+	defer test_helpers.StopTarantoolWithCleanup(inst)
+
+	conn = test_helpers.ConnectWithValidation(t, shtdnServer, shtdnClntOpts)
+	defer conn.Close()
+
+	// Send request with sleep.
+	evalSleep := 3 // in seconds
+	require.Lessf(t,
+		time.Duration(evalSleep)*time.Second,
+		shtdnClntOpts.Timeout,
+		"test request won't be failed by timeout")
+
+	req := NewEvalRequest(evalBody).Args([]interface{}{evalSleep, evalMsg})
+	fut := conn.Do(req)
+
+	// Close the connection.
+	conn.Close()
+
+	// Connection is closed.
+	require.Equal(t, true, conn.ClosedNow())
+
+	// Check that request was successful.
+	resp, err := fut.Get()
+	require.Nilf(t, err, "sleep request no error")
+	require.NotNilf(t, resp, "sleep response exists")
+}
+
 func TestGracefulShutdownWithReconnect(t *testing.T) {
 	test_helpers.SkipIfWatchersUnsupported(t)
 
@@ -222,86 +258,6 @@ func TestNoGracefulShutdown(t *testing.T) {
 		"server went down without any additional waiting")
 }
 
-func TestGracefulShutdownRespectsClose(t *testing.T) {
-	test_helpers.SkipIfWatchersUnsupported(t)
-
-	var inst test_helpers.TarantoolInstance
-	var conn *Connection
-	var err error
-
-	inst, err = test_helpers.StartTarantool(shtdnSrvOpts)
-	require.Nil(t, err)
-	defer test_helpers.StopTarantoolWithCleanup(inst)
-
-	conn = test_helpers.ConnectWithValidation(t, shtdnServer, shtdnClntOpts)
-	defer conn.Close()
-
-	// Create a helper watcher to ensure that async
-	// shutdown is set up.
-	helperCh := make(chan WatchEvent, 10)
-	helperW, herr := conn.NewWatcher("box.shutdown", func(event WatchEvent) {
-		helperCh <- event
-	})
-	require.Nil(t, herr)
-	defer helperW.Unregister()
-	<-helperCh
-
-	// Set a big timeout so it would be easy to differ
-	// if server went down on timeout or after all connections were terminated.
-	serverShutdownTimeout := 60 // in seconds
-	_, err = conn.Call("box.ctl.set_on_shutdown_timeout", []interface{}{serverShutdownTimeout})
-	require.Nil(t, err)
-
-	// Send request with sleep.
-	evalSleep := 10 // in seconds
-	require.Lessf(t,
-		time.Duration(evalSleep)*time.Second,
-		shtdnClntOpts.Timeout,
-		"test request won't be failed by timeout")
-
-	req := NewEvalRequest(evalBody).Args([]interface{}{evalSleep, evalMsg})
-
-	fut := conn.Do(req)
-
-	// SIGTERM the server.
-	shutdownStart := time.Now()
-	require.Nil(t, inst.Cmd.Process.Signal(syscall.SIGTERM))
-
-	// Close the connection.
-	conn.Close()
-
-	// Connection is closed.
-	require.Equal(t, true, conn.ClosedNow())
-
-	// Check that request was interrupted.
-	_, err = fut.Get()
-	require.NotNilf(t, err, "sleep request error")
-
-	// Wait until server go down.
-	_, err = inst.Cmd.Process.Wait()
-	require.Nil(t, err)
-	shutdownFinish := time.Now()
-	shutdownTime := shutdownFinish.Sub(shutdownStart)
-
-	// Help test helpers to properly clean up.
-	inst.Cmd.Process = nil
-
-	// Check that server finished without waiting for eval to finish.
-	require.Lessf(t,
-		shutdownTime,
-		time.Duration(evalSleep/2)*time.Second,
-		"server went down without any additional waiting")
-
-	// Check that it wasn't a timeout.
-	require.Lessf(t,
-		shutdownTime,
-		time.Duration(serverShutdownTimeout/2)*time.Second,
-		"server went down not by timeout")
-
-	// Connection is still closed.
-	require.Equal(t, true, conn.ClosedNow())
-}
-
 func TestGracefulShutdownNotRacesWithRequestReconnect(t *testing.T) {
 	test_helpers.SkipIfWatchersUnsupported(t)