MXS-1503: Enable master reconnection

The `master_reconnection` parameter now controls both the reconnection of
the master server as well as the migration of the master server to another
server. Although these two cases appear to be different, the end result
from readwritesplit's point of view is the same and are thus controlled
with the same parameter.

The RWBackend class now resets its internal state when it is closed. This
allows readwritesplit to handle the case when a result was expected from
the master but the master died before the result was returned. The same
code should also handle slave connection failures mid-result, allowing
Backend reuse.

Added a test case that verifies the new functionality when combined with
`master_failure_mode=error_on_write`.

server/modules/routing/readwritesplit/readwritesplit.cc

@@ -1376,17 +1376,32 @@ static void handleError(MXS_ROUTER *instance,
                         can_continue = true;
                     }
                 }
-                else if (!SERVER_IS_MASTER(srv) && !srv->master_err_is_logged)
+                else
                 {
-                    ss_dassert(backend);
-                    MXS_ERROR("Server %s (%s) lost the master status while waiting"
-                              " for a result. Client sessions will be closed.",
-                              backend->name(), backend->uri());
-                    backend->server()->master_err_is_logged = true;
+                    // We were expecting a response but we aren't going to get one
+                    rses->expected_responses--;
+
+                    if (rses->rses_config.master_failure_mode == RW_ERROR_ON_WRITE)
+                    {
+                        /** In error_on_write mode, the session can continue even
+                         * if the master is lost. Send a read-only error to
+                         * the client to let it know that the query failed. */
+                        can_continue = true;
+                        send_readonly_error(rses->client_dcb);
+                    }
+
+                    if (!SERVER_IS_MASTER(srv) && !srv->master_err_is_logged)
+                    {
+                        ss_dassert(backend);
+                        MXS_ERROR("Server %s (%s) lost the master status while waiting"
+                            " for a result. Client sessions will be closed.",
+                                  backend->name(), backend->uri());
+                        backend->server()->master_err_is_logged = true;
+                    }
                 }
 
                 *succp = can_continue;
-                backend->close(mxs::Backend::CLOSE_FATAL);
+                backend->close();
             }
             else
             {

server/modules/routing/readwritesplit/rwsplit_route_stmt.cc

@@ -167,6 +167,30 @@ bool route_single_stmt(RWSplit *inst, RWSplitSession *rses, GWBUF *querybuf, con
         {
             succp = handle_master_is_target(inst, rses, &target);
 
+            // Check if we need to connect to the master server in order to use it
+            if (succp && target && !target->in_use() && target->can_connect())
+            {
+                if (rses->rses_config.master_reconnection)
+                {
+                    if ((!rses->rses_config.disable_sescmd_history || rses->recv_sescmd == 0))
+                    {
+                        target->connect(rses->client_dcb->session, &rses->sescmd_list);
+                    }
+                    else
+                    {
+                        MXS_ERROR("Cannot reconnect to master, session command"
+                                  " history is disabled (session has executed"
+                                  " %lu session commands).", rses->recv_sescmd);
+                    }
+                }
+                else
+                {
+                    MXS_ERROR("The connection to the master was lost but "
+                              "'master_reconnection' is not enabled.");
+                    succp = false;
+                }
+            }
+
             if (!rses->rses_config.strict_multi_stmt &&
                 !rses->rses_config.strict_sp_calls &&
                 rses->target_node == rses->current_master)
@@ -462,7 +486,7 @@ SRWBackend get_target_backend(RWSplitSession *rses, backend_type_t btype,
 
         if (master)
         {
-            if (master->in_use())
+            if (master->in_use() || master->can_connect())
             {
                 if (master->is_master())
                 {
@@ -480,10 +504,6 @@ SRWBackend get_target_backend(RWSplitSession *rses, backend_type_t btype,
                           master->name());
             }
         }
-        else
-        {
-            MXS_ERROR("No master server available at this time.");
-        }
     }
 
     return rval;

server/modules/routing/readwritesplit/rwsplit_select_backends.cc

@@ -351,12 +351,9 @@ bool select_connect_backend_servers(RWSplit *inst, MXS_SESSION *session,
     {
         if (backend->can_connect() && backend->connect(session, sescmd_list))
         {
-            if (sescmd_list && sescmd_list->size())
+            if (sescmd_list && sescmd_list->size() && expected_responses)
             {
-                if (expected_responses)
-                {
-                    (*expected_responses)++;
-                }
+                (*expected_responses)++;
             }
 
             slaves_connected++;

server/modules/routing/readwritesplit/rwsplitsession.cc

@@ -79,6 +79,12 @@ bool RWBackend::write(GWBUF* buffer, response_type type)
     return mxs::Backend::write(buffer);
 }
 
+void RWBackend::close(close_type type)
+{
+    m_reply_state = REPLY_STATE_DONE;
+    mxs::Backend::close(type);
+}
+
 uint32_t get_internal_ps_id(RWSplitSession* rses, GWBUF* buffer)
 {
     uint32_t rval = 0;

server/modules/routing/readwritesplit/rwsplitsession.hh

@@ -66,6 +66,7 @@ public:
 
     bool execute_session_command();
     bool write(GWBUF* buffer, response_type type = EXPECT_RESPONSE);
+    void close(close_type type = CLOSE_NORMAL);
 
     inline void set_large_packet(bool value)
     {