MXS-1810: Create Trx class

The class encapsulates the relevant information of a transaction.

server/modules/routing/readwritesplit/rwsplit_route_stmt.cc

@@ -235,19 +235,22 @@ bool RWSplitSession::route_single_stmt(GWBUF *querybuf)
                 // Target server was found and is in the correct state
                 succp = handle_got_target(querybuf, target, store_stmt);
 
-                if (succp && command == MXS_COM_STMT_EXECUTE && not_locked_to_master)
+                if (succp)
                 {
-                    /** Track the targets of the COM_STMT_EXECUTE statements. This
+                    if (command == MXS_COM_STMT_EXECUTE && not_locked_to_master)
-                     * information is used to route all COM_STMT_FETCH commands
+                    {
-                     * to the same server where the COM_STMT_EXECUTE was done. */
+                        /** Track the targets of the COM_STMT_EXECUTE statements. This
-                    ss_dassert(stmt_id > 0);
+                         * information is used to route all COM_STMT_FETCH commands
-                    m_exec_map[stmt_id] = target;
+                         * to the same server where the COM_STMT_EXECUTE was done. */
-                    MXS_INFO("COM_STMT_EXECUTE on %s: %s", target->name(), target->uri());
+                        ss_dassert(stmt_id > 0);
-                }
+                        m_exec_map[stmt_id] = target;
+                        MXS_INFO("COM_STMT_EXECUTE on %s: %s", target->name(), target->uri());
+                    }
 
-                if (succp && session_trx_is_active(m_client->session))
+                    if (session_trx_is_active(m_client->session))
-                {
+                    {
-                    m_trx_log.push_back(gwbuf_clone(querybuf));
+                        m_trx.add_stmt(gwbuf_clone(querybuf));
+                    }
                 }
             }
         }

server/modules/routing/readwritesplit/rwsplitsession.cc

@@ -407,7 +407,7 @@ void RWSplitSession::clientReply(GWBUF *writebuf, DCB *backend_dcb)
     }
     else if (session_trx_is_active(m_client->session))
     {
-        m_trx_checksum.update(writebuf);
+        m_trx.add_result(writebuf);
     }
 
     if (backend->reply_is_complete(writebuf))
@@ -763,14 +763,14 @@ bool RWSplitSession::supports_hint(HINT_TYPE hint_type) const
 
 void RWSplitSession::close_transaction()
 {
-    m_trx_checksum.finalize();
+    m_trx.finalize();
-    MXS_INFO("Checksum of current transaction: %s", m_trx_checksum.hex().c_str());
+    MXS_INFO("Checksum of current transaction: %s", m_trx.checksum().hex().c_str());
     int i = 1;
 
-    while (!m_trx_log.empty())
+    while (!m_trx.empty())
     {
         const int max_len = 1024;
-        MXS_INFO("%d: %s", i++, mxs::extract_sql(m_trx_log.front().get(), max_len).c_str());
+        mxs::Buffer buf = m_trx.pop_stmt();
-        m_trx_log.pop_front();
+        MXS_INFO("%d: %s", i++, mxs::extract_sql(buf.get(), max_len).c_str());
     }
 }

server/modules/routing/readwritesplit/rwsplitsession.hh

@@ -14,13 +14,12 @@
 
 #include "readwritesplit.hh"
 #include "rwbackend.hh"
+#include "trx.hh"
 
 #include <string>
-#include <deque>
 
 #include <maxscale/buffer.hh>
 #include <maxscale/modutil.h>
-#include <maxscale/utils.hh>
 #include <maxscale/queryclassifier.hh>
 
 #define TARGET_IS_MASTER(t)         maxscale::QueryClassifier::target_is_master(t)
@@ -40,9 +39,6 @@ typedef std::list< std::pair<mxs::SRWBackend, uint8_t> > SlaveResponseList;
 /** Map of COM_STMT_EXECUTE targets by internal ID */
 typedef std::tr1::unordered_map<uint32_t, mxs::SRWBackend> ExecMap;
 
-// A log of executed queries, for transaction replay
-typedef std::deque<mxs::Buffer> TrxLog;
-
 /**
  * The client session of a RWSplit instance
  */
@@ -141,8 +137,7 @@ public:
     mxs::QueryClassifier    m_qc; /**< The query classifier. */
     uint64_t                m_retry_duration; /**< Total time spent retrying queries */
     mxs::Buffer             m_current_query; /**< Current query being executed */
-    mxs::SHA1Checksum       m_trx_checksum; /**< Transaction checksum */
+    Trx                     m_trx; /**< Current transaction */
-    TrxLog                  m_trx_log; /**< Log of executed queries in the current transaction */
 
 private:
     RWSplitSession(RWSplit* instance, MXS_SESSION* session,

server/modules/routing/readwritesplit/trx.hh

@@ -0,0 +1,104 @@
+#pragma once
+/*
+ * Copyright (c) 2016 MariaDB Corporation Ab
+ *
+ * Use of this software is governed by the Business Source License included
+ * in the LICENSE.TXT file and at www.mariadb.com/bsl11.
+ *
+ * Change Date: 2020-01-01
+ *
+ * On the date above, in accordance with the Business Source License, use
+ * of this software will be governed by version 2 or later of the General
+ * Public License.
+ */
+
+#include <maxscale/cppdefs.hh>
+
+#include <deque>
+
+#include <maxscale/buffer.hh>
+#include <maxscale/utils.hh>
+
+// A log of executed queries, for transaction replay
+typedef std::deque<mxs::Buffer> TrxLog;
+
+// A transaction
+class Trx
+{
+public:
+
+    /**
+     * Add a statement to the transaction
+     *
+     * @param buf Statement to add
+     */
+    void add_stmt(GWBUF* buf)
+    {
+        m_log.push_back(buf);
+    }
+
+    /**
+     * Add a result to the transaction
+     *
+     * The result is used to update the checksum.
+     *
+     * @param buf Result to add
+     */
+    void add_result(GWBUF* buf)
+    {
+        m_checksum.update(buf);
+    }
+
+    /**
+     * Releases the oldest statement in this transaction
+     *
+     * This reduces the size of the transaction by one and should only be used
+     * to replay a transaction.
+     *
+     * @return The oldest statement in this transaction
+     */
+    mxs::Buffer pop_stmt()
+    {
+        mxs::Buffer rval = m_log.front();
+        m_log.pop_front();
+        return rval;
+    }
+
+    /**
+     * Finalize the transaction
+     *
+     * This function marks the transaction as completed be that by a COMMIT
+     * or by a failure of the current server where the transaction was being
+     * executed.
+     */
+    void finalize()
+    {
+        m_checksum.finalize();
+    }
+
+    /**
+     * Check if transaction is empty
+     *
+     * @return True if transaction has no statements
+     */
+    bool empty() const
+    {
+        return m_log.empty();
+    }
+
+    /**
+     * Return the current checksum
+     *
+     * finalize() must be called before the return value of this function is used.
+     *
+     * @return The checksum of the transaction
+     */
+    const mxs::SHA1Checksum& checksum() const
+    {
+        return m_checksum;
+    }
+
+private:
+    mxs::SHA1Checksum m_checksum; /**< Checksum of the transaction */
+    TrxLog            m_log; /**< The transaction contents */
+} ;