MXS-852: Add PS manager class

The class manages both text and binary protocol prepared statement ID to
type mapping. The text protocol statements are mapped by their plaintext
name and the binary protocol statements are mapped by the session command
ID of the prepared statement.

By mapping the binary protocol prepared statement type to the session
command identifier, we can store the types for both styles of prepared
statements in a very similar manner. When the prepared statement handle is
received from the backend and is sent to the client, the client handle to
session command ID mapping can be done. This allows the mapping of both
client and backend PS handles to internal session command IDs.

server/modules/routing/readwritesplit/readwritesplit.hh

@@ -217,15 +217,58 @@ private:
     HandleMap     m_ps_handles;
 };
 
+/** Prepared statement ID to type maps for text protocols */
+typedef std::tr1::unordered_map<uint64_t, uint32_t>    BinaryPSMap;
+typedef std::tr1::unordered_map<std::string, uint32_t> TextPSMap;
+
+class PSManager
+{
+    PSManager(const PSManager&);
+    PSManager& operator =(const PSManager&);
+
+public:
+    PSManager();
+    ~PSManager();
+
+    /**
+     * @brief Store and process a prepared statement
+     *
+     * @param buffer Buffer containing either a text or a binary protocol
+     *               prepared statement
+     * @param id     The unique ID for this statement
+     */
+    void store(GWBUF* buffer, uint64_t id);
+
+    /**
+     * @brief Get the type of a stored prepared statement
+     *
+     * @param id The unique identifier for the prepared statement or the plaintext
+     *           name of the prepared statement
+     *
+     * @return The type of the prepared statement
+     */
+    uint32_t get_type(uint64_t id) const;
+    uint32_t get_type(std::string id) const;
+
+    /**
+     * @brief Remove a prepared statement
+     *
+     * @param id Statement identifier to remove
+     */
+    void erase(std::string id);
+    void erase(uint64_t id);
+
+private:
+    BinaryPSMap m_binary_ps;
+    TextPSMap   m_text_ps;
+};
+
 typedef std::tr1::shared_ptr<RWBackend> SRWBackend;
 typedef std::list<SRWBackend> SRWBackendList;
 
 typedef std::tr1::unordered_set<std::string> TableSet;
 typedef std::map<uint64_t, uint8_t>          ResponseMap;
 
-/** Prepared statement ID to type maps for text and binary protocols */
-typedef std::tr1::unordered_map<std::string, uint32_t> TextPSMap;
-
 /**
  * The client session structure used within this router.
  */
@@ -252,7 +295,7 @@ struct ROUTER_CLIENT_SES
     ResponseMap               sescmd_responses; /**< Response to each session command */
     uint64_t                  sent_sescmd; /**< ID of the last sent session command*/
     uint64_t                  recv_sescmd; /**< ID of the most recently completed session command */
-    TextPSMap                 ps_text;     /**< Text protocol prepared statements */
+    PSManager                 ps_manager;   /**< Prepared statement manager*/
     skygw_chk_t               rses_chk_tail;
 };
 

server/modules/routing/readwritesplit/rwsplit_internal.hh

@@ -76,7 +76,8 @@ bool handle_master_is_target(ROUTER_INSTANCE *inst, ROUTER_CLIENT_SES *rses,
                              SRWBackend* dest);
 bool handle_got_target(ROUTER_INSTANCE *inst, ROUTER_CLIENT_SES *rses,
                        GWBUF *querybuf, SRWBackend& target, bool store);
-bool route_session_write(ROUTER_CLIENT_SES *rses, GWBUF *querybuf, uint8_t command);
+bool route_session_write(ROUTER_CLIENT_SES *rses, GWBUF *querybuf,
+                         uint8_t command, uint32_t type);
 
 void process_sescmd_response(ROUTER_CLIENT_SES* rses, SRWBackend& bref,
                              GWBUF** ppPacket, bool* reconnect);
@@ -114,9 +115,10 @@ uint32_t determine_query_type(GWBUF *querybuf, int packet_type, bool non_empty_p
 void close_all_connections(ROUTER_CLIENT_SES* rses);
 
 /**
- * Functions for prepared statement handling
+ * @brief Extract text identifier of a PREPARE or EXECUTE statement
+ *
+ * @param buffer Buffer containing a PREPARE or EXECUTE command
+ *
+ * @return The string identifier of the statement
  */
 std::string extract_text_ps_id(GWBUF* buffer);
-void store_text_ps(ROUTER_CLIENT_SES* rses, std::string id, GWBUF* buffer);
-void erase_text_ps(ROUTER_CLIENT_SES* rses, std::string id);
-bool get_text_ps_type(ROUTER_CLIENT_SES* rses, GWBUF* buffer, uint32_t* out);

server/modules/routing/readwritesplit/rwsplit_mysql.cc

@@ -207,15 +207,6 @@ bool handle_target_is_all(route_target_t route_target, ROUTER_INSTANCE *inst,
 {
     bool result = false;
 
-    if (qc_query_is_type(qtype, QUERY_TYPE_PREPARE_NAMED_STMT))
-    {
-        store_text_ps(rses, extract_text_ps_id(querybuf), querybuf);
-    }
-    else if (qc_query_is_type(qtype, QUERY_TYPE_PREPARE_STMT))
-    {
-        gwbuf_set_type(querybuf, GWBUF_TYPE_COLLECT_RESULT);
-    }
-
     if (TARGET_IS_MASTER(route_target) || TARGET_IS_SLAVE(route_target))
     {
         /**
@@ -243,7 +234,7 @@ bool handle_target_is_all(route_target_t route_target, ROUTER_INSTANCE *inst,
         MXS_FREE(query_str);
         MXS_FREE(qtype_str);
     }
-    else if (route_session_write(rses, gwbuf_clone(querybuf), packet_type))
+    else if (route_session_write(rses, gwbuf_clone(querybuf), packet_type, qtype))
     {
 
         result = true;

server/modules/routing/readwritesplit/rwsplit_ps.cc

@@ -15,6 +15,46 @@
 
 #include <maxscale/alloc.h>
 #include <maxscale/query_classifier.h>
+#include <maxscale/protocol/mysql.h>
+
+static uint32_t get_prepare_type(GWBUF* buffer)
+{
+    uint32_t type;
+
+    if (mxs_mysql_get_command(buffer) == MYSQL_COM_STMT_PREPARE)
+    {
+        // TODO: This could be done inside the query classifier
+        size_t packet_len = gwbuf_length(buffer);
+        size_t payload_len = packet_len - MYSQL_HEADER_LEN;
+        GWBUF* stmt = gwbuf_alloc(packet_len);
+        uint8_t* ptr = GWBUF_DATA(stmt);
+
+        // Payload length
+        *ptr++ = payload_len;
+        *ptr++ = (payload_len >> 8);
+        *ptr++ = (payload_len >> 16);
+        // Sequence id
+        *ptr++ = 0x00;
+        // Command
+        *ptr++ = MYSQL_COM_QUERY;
+
+        gwbuf_copy_data(buffer, MYSQL_HEADER_LEN + 1, payload_len - 1, ptr);
+        type = qc_get_type_mask(stmt);
+
+        gwbuf_free(stmt);
+    }
+    else
+    {
+        GWBUF* stmt = qc_get_preparable_stmt(buffer);
+        ss_dassert(stmt);
+
+        type = qc_get_type_mask(stmt);
+    }
+
+    ss_dassert((type & (QUERY_TYPE_PREPARE_STMT | QUERY_TYPE_PREPARE_NAMED_STMT)) == 0);
+
+    return type;
+}
 
 std::string extract_text_ps_id(GWBUF* buffer)
 {
@@ -30,39 +70,83 @@ std::string extract_text_ps_id(GWBUF* buffer)
     return rval;
 }
 
-void store_text_ps(ROUTER_CLIENT_SES* rses, std::string id, GWBUF* buffer)
+PSManager::PSManager()
 {
-    GWBUF* stmt = qc_get_preparable_stmt(buffer);
-    ss_dassert(stmt);
-
-    uint32_t type = qc_get_type_mask(stmt);
-    ss_dassert((type & (QUERY_TYPE_PREPARE_STMT | QUERY_TYPE_PREPARE_NAMED_STMT)) == 0);
-
-    rses->ps_text[id] = type;
 }
 
-void erase_text_ps(ROUTER_CLIENT_SES* rses, std::string id)
+PSManager::~PSManager()
 {
-    rses->ps_text.erase(id);
 }
 
-bool get_text_ps_type(ROUTER_CLIENT_SES* rses, GWBUF* buffer, uint32_t* out)
+void PSManager::erase(uint64_t id)
 {
-    bool rval = false;
+    if (m_binary_ps.erase(id) == 0)
-    char* name = qc_get_prepare_name(buffer);
-
-    if (name)
     {
-        TextPSMap::iterator it = rses->ps_text.find(name);
+        MXS_WARNING("Closing unknown prepared statement with ID %lu", id);
-
-        if (it != rses->ps_text.end())
-        {
-            *out = it->second;
-            rval = true;
     }
+}
 
-        MXS_FREE(name);
+void PSManager::erase(std::string id)
+{
+    if (m_text_ps.erase(id) == 0)
+    {
+        MXS_WARNING("Closing unknown prepared statement with ID '%s'", id.c_str());
+    }
+}
+
+uint32_t PSManager::get_type(std::string id) const
+{
+    uint32_t rval = QUERY_TYPE_UNKNOWN;
+    TextPSMap::const_iterator it = m_text_ps.find(id);
+
+    if (it != m_text_ps.end())
+    {
+        rval = it->second;
+    }
+    else
+    {
+        MXS_WARNING("Using unknown prepared statement with ID '%s'", id.c_str());
     }
 
     return rval;
 }
+
+
+uint32_t PSManager::get_type(uint64_t id) const
+{
+    uint32_t rval = QUERY_TYPE_UNKNOWN;
+    BinaryPSMap::const_iterator it = m_binary_ps.find(id);
+
+    if (it != m_binary_ps.end())
+    {
+        rval = it->second;
+    }
+    else
+    {
+        MXS_WARNING("Using unknown prepared statement with ID %lu", id);
+    }
+
+    return rval;
+}
+
+void PSManager::store(GWBUF* buffer, uint64_t id)
+{
+    ss_dassert(mxs_mysql_get_command(buffer) == MYSQL_COM_STMT_PREPARE ||
+               qc_query_is_type(qc_get_type_mask(buffer),
+                                QUERY_TYPE_PREPARE_NAMED_STMT));
+
+    switch (mxs_mysql_get_command(buffer))
+    {
+    case MYSQL_COM_QUERY:
+        m_text_ps[extract_text_ps_id(buffer)] = get_prepare_type(buffer);
+        break;
+
+    case MYSQL_COM_STMT_PREPARE:
+        m_binary_ps[id] = get_prepare_type(buffer);
+        break;
+
+    default:
+        ss_dassert(!true);
+        break;
+    }
+}

server/modules/routing/readwritesplit/rwsplit_route_stmt.cc

@@ -142,10 +142,11 @@ bool route_single_stmt(ROUTER_INSTANCE *inst, ROUTER_CLIENT_SES *rses,
 
         uint32_t ps_type;
 
-        if (qc_get_operation(querybuf) == QUERY_OP_EXECUTE &&
+        if (command == MYSQL_COM_QUERY &&
-            get_text_ps_type(rses, querybuf, &ps_type))
+            qc_get_operation(querybuf) == QUERY_OP_EXECUTE)
         {
-            qtype = ps_type;
+            std::string id = extract_text_ps_id(querybuf);
+            qtype = rses->ps_manager.get_type(id);
         }
 
         route_target = get_route_target(rses, qtype, querybuf->hint);
@@ -237,7 +238,8 @@ bool route_single_stmt(ROUTER_INSTANCE *inst, ROUTER_CLIENT_SES *rses,
  * backends being used, otherwise false.
  *
  */
-bool route_session_write(ROUTER_CLIENT_SES *rses, GWBUF *querybuf, uint8_t command)
+bool route_session_write(ROUTER_CLIENT_SES *rses, GWBUF *querybuf,
+                         uint8_t command, uint32_t type)
 {
     /** The SessionCommand takes ownership of the buffer */
     uint64_t id = rses->sescmd_count++;
@@ -246,6 +248,13 @@ bool route_session_write(ROUTER_CLIENT_SES *rses, GWBUF *querybuf, uint8_t comma
     int nsucc = 0;
     uint64_t lowest_pos = id;
 
+    if (qc_query_is_type(type, QUERY_TYPE_PREPARE_NAMED_STMT) ||
+        qc_query_is_type(type, QUERY_TYPE_PREPARE_STMT))
+    {
+        gwbuf_set_type(querybuf, GWBUF_TYPE_COLLECT_RESULT);
+        rses->ps_manager.store(querybuf, id);
+    }
+
     MXS_INFO("Session write, routing to all servers.");
 
     for (SRWBackendList::iterator it = rses->backends.begin();