reconnect works

Author: DifferentialOrange

connection.go

@@ -454,26 +454,30 @@ func Connect(addr string, opts Opts) (conn *Connection, err error) {
 
 func (conn *Connection) shutdownOnServerEvent() {
 	conn.mutex.Lock()
-	defer conn.mutex.Unlock()
 
-	fmt.Printf("starting shutdown...\n")
+	fmt.Printf("starting shutdown... (%v)\n", conn.UniqueId())
 	conn.shutdownOn.Store(true)
 	defer conn.shutdownOn.Store(false)
 
-	fmt.Printf("wait for futures to finish...\n")
+	fmt.Printf("wait for futures to finish... (%v)\n", conn.UniqueId())
 	conn.shutdownWg.Wait()
 
 	// fmt.Printf("defer unregister watcher...\n")
 	// defer conn.shutdownWatcher.Unregister()
 
-	fmt.Printf("close the connection...\n")
-	conn.closeConnection(
-		ClientError{
-			ErrConnectionClosed,
-			"connection closed after server shutdown",
-		}, false)
+	err := ClientError{
+		ErrConnectionClosed,
+		"connection closed after server shutdown",
+	}
+
+	fmt.Printf("close the connection... (%v)\n", conn.UniqueId())
+	conn.closeConnection(err, false)
+
+	fmt.Printf("finish shutdown... (%v)\n", conn.UniqueId())
 
-	fmt.Printf("finish shutdown...\n")
+	conn.mutex.Unlock()
+
+	defer conn.reconnect(err, conn.c)
 }
 
 // ConnectedNow reports if connection is established at the moment.
@@ -779,7 +783,10 @@ func (conn *Connection) readIdResponse(r io.Reader) (Response, error) {
 
 func (conn *Connection) createConnection(reconnect bool) (err error) {
 	var reconnects uint
+	fmt.Printf("conn.state is %v (%v)\n", conn.state, conn.UniqueId())
 	for conn.c == nil && conn.state == connDisconnected {
+		fmt.Printf("createConnection iteration (%v)\n", conn.UniqueId())
+
 		now := time.Now()
 		err = conn.dial()
 		if err == nil || !reconnect {
@@ -797,9 +804,12 @@ func (conn *Connection) createConnection(reconnect bool) (err error) {
 		conn.opts.Logger.Report(LogReconnectFailed, conn, reconnects, err)
 		conn.notify(ReconnectFailed)
 		reconnects++
+
+		fmt.Printf("until sleep %v (%v)\n", time.Now(), conn.UniqueId())
 		conn.mutex.Unlock()
 		time.Sleep(time.Until(now.Add(conn.opts.Reconnect)))
 		conn.mutex.Lock()
+		fmt.Printf("after sleep %v (%v)\n", time.Now(), conn.UniqueId())
 	}
 	if conn.state == connClosed {
 		err = ClientError{ErrConnectionClosed, "using closed connection"}
@@ -836,19 +846,29 @@ func (conn *Connection) closeConnection(neterr error, forever bool) (err error)
 	return
 }
 
+func (conn *Connection) UniqueId() string {
+	return fmt.Sprintf("conn %v", conn.control)
+}
+
 func (conn *Connection) reconnect(neterr error, c net.Conn) {
+	fmt.Printf("reconnect head (%v)\n", conn.UniqueId())
 	conn.mutex.Lock()
 	defer conn.mutex.Unlock()
+	fmt.Printf("conn.opts.Reconnect %d (%v)\n", conn.opts.Reconnect, conn.UniqueId())
 	if conn.opts.Reconnect > 0 {
 		if c == conn.c {
+			fmt.Printf("reconnect attempt (%v)\n", conn.UniqueId())
 			conn.closeConnection(neterr, false)
 			if err := conn.createConnection(true); err != nil {
+				fmt.Printf("kill that one (folded)(%v)\n", conn.UniqueId())
 				conn.closeConnection(err, true)
 			}
 		}
 	} else {
+		fmt.Printf("kill that one (%v)\n", conn.UniqueId())
 		conn.closeConnection(neterr, true)
 	}
+	fmt.Printf("reconnect end (%v)\n", conn.UniqueId())
 }
 
 func (conn *Connection) lockShards() {
@@ -1166,11 +1186,11 @@ func (conn *Connection) send(req Request, streamId uint64) *Future {
 }
 
 func (conn *Connection) gracefulWait(fut *Future) {
-	fmt.Printf("start tracking %v\n", fut)
+	// fmt.Printf("start tracking %v\n", fut)
 	<-fut.done
-	fmt.Printf("done tracking %v\n", fut)
+	// fmt.Printf("done tracking %v\n", fut)
 	conn.shutdownWg.Done()
-	fmt.Printf("wg done %v\n", fut)
+	// fmt.Printf("wg done %v\n", fut)
 }
 
 func (conn *Connection) putFuture(fut *Future, req Request, streamId uint64) {
@@ -1619,7 +1639,7 @@ func (conn *Connection) NewWatcher(key string, callback WatchCallback) (Watcher,
 
 				if sendAck {
 					fut := conn.Do(newWatchRequest(key))
-					fmt.Printf("ack request %v %v: %v\n", key, state.value, fut)
+					// fmt.Printf("ack request %v %v: %v\n", key, state.value, fut)
 					fut.Get()
 					// We expect a reconnect and re-subscribe if it fails to
 					// send the watch request. So it looks ok do not check a

graceful_shutdown/tarantool_test.go

@@ -45,7 +45,10 @@ func TestGracefulShutdown(t *testing.T) {
 	inst, err := test_helpers.StartTarantool(instOpts)
 	require.Nil(t, err)
 
+	// fmt.Printf("opts are is %v\n", opts)
+	// fmt.Printf("opts.Reconnect is %v\n", opts.Reconnect)
 	conn := test_helpers.ConnectWithValidation(t, server, opts)
+	fmt.Printf("test connect unique id: %v\n", conn.UniqueId())
 
 	_, err = conn.Eval("box.ctl.set_on_shutdown_timeout(10)", []interface{}{})
 	require.Nil(t, err)
@@ -93,6 +96,7 @@ func TestGracefulShutdown(t *testing.T) {
 	err = test_helpers.RestartTarantool(&inst)
 	require.Nilf(t, err, "Failed to restart tarantool")
 
+	fmt.Printf("opts are is %v\n", opts)
 	connected := test_helpers.WaitUntilReconnected(conn, retries, timeout)
 	require.True(t, connected, "Reconnect success")
 
@@ -125,5 +129,5 @@ func TestGracefulShutdown(t *testing.T) {
 
 	// time.Sleep(2 * time.Second)
 
-	require.Equal(t, true, conn.ClosedNow())
+	require.Equal(t, false, conn.ConnectedNow())
 }

test_helpers/utils.go

@@ -47,9 +47,14 @@ func DeleteRecordByKey(t *testing.T, conn tarantool.Connector,
 // Returns false in case of connection is not in the connected state
 // after specified retries count, true otherwise.
 func WaitUntilReconnected(conn *tarantool.Connection, retries uint, timeout time.Duration) bool {
+	fmt.Printf("WaitUntilReconnected %v %v \n", retries, timeout)
+
 	for i := uint(0); ; i++ {
+		fmt.Printf("reconnect in tests iteration i=%v\n", i)
+
 		connected := conn.ConnectedNow()
 		if connected {
+			fmt.Printf("ConnectedNow true\n")
 			return true
 		}
 
@@ -58,6 +63,10 @@ func WaitUntilReconnected(conn *tarantool.Connection, retries uint, timeout time
 		}
 
 		time.Sleep(timeout)
+
+		// resp, err := conn.Do(tarantool.NewPingRequest()).Get()
+		fmt.Printf("reconnect in tests iteration i=%v\n", i)
+		// fmt.Printf("reconnect tests; err ping %v\n", err)
 	}
 
 	return false