mysql_connection.h

Go to the documentation of this file.

// Copyright (C) 2012-2024 Internet Systems Consortium, Inc. ("ISC")
//
// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.
 
#ifndef MYSQL_CONNECTION_H
#define MYSQL_CONNECTION_H
 
#include <asiolink/io_service.h>
#include <database/database_connection.h>
#include <database/db_exceptions.h>
#include <database/db_log.h>
#include <exceptions/exceptions.h>
#include <mysql/mysql_binding.h>
#include <mysql/mysql_constants.h>
#include <boost/scoped_ptr.hpp>
#include <mysql.h>
#include <mysqld_error.h>
#include <errmsg.h>
#include <functional>
#include <vector>
#include <stdint.h>
 
namespace isc {
namespace db {
 
 
 
class MySqlFreeResult {
public:
 
    MySqlFreeResult(MYSQL_STMT* statement) : statement_(statement)
    {}
 
    ~MySqlFreeResult() {
        (void) mysql_stmt_free_result(statement_);
    }
 
private:
    MYSQL_STMT*     statement_;     
};
 
struct TaggedStatement {
    uint32_t index;
    const char* text;
};
 
template <typename Fun, typename... Args>
int retryOnDeadlock(Fun& fun, Args... args) {
    int status;
    for (unsigned count = 0; count < 5; ++count) {
        status = fun(args...);
        if (status != ER_LOCK_DEADLOCK) {
            break;
        }
    }
    return (status);
}
 
inline int MysqlExecuteStatement(MYSQL_STMT* stmt) {
    return (retryOnDeadlock(mysql_stmt_execute, stmt));
}
 
inline int MysqlQuery(MYSQL* mysql, const char* stmt) {
    return (retryOnDeadlock(mysql_query, mysql, stmt));
}
 
class MySqlHolder : public boost::noncopyable {
public:
 
    MySqlHolder() : mysql_(mysql_init(NULL)) {
        if (mysql_ == NULL) {
            isc_throw(db::DbOpenError, "unable to initialize MySQL");
        }
    }
 
    ~MySqlHolder() {
        if (mysql_ != NULL) {
            mysql_close(mysql_);
        }
    }
 
    operator MYSQL*() const {
        return (mysql_);
    }
 
private:
    static int atexit_;
 
    MYSQL* mysql_;
};
 
class MySqlConnection;
 
class MySqlTransaction : public boost::noncopyable {
public:
 
    MySqlTransaction(MySqlConnection& conn);
 
    ~MySqlTransaction();
 
    void commit();
 
private:
 
    MySqlConnection& conn_;
 
    bool committed_;
};
 
 
class MySqlConnection : public db::DatabaseConnection {
public:
 
    typedef std::function<void(MySqlBindingCollection&)> ConsumeResultFun;
 
    MySqlConnection(const ParameterMap& parameters,
                    IOServiceAccessorPtr io_accessor = IOServiceAccessorPtr(),
                    DbCallback callback = DbCallback())
        : DatabaseConnection(parameters, callback),
          io_service_accessor_(io_accessor), io_service_(),
          transaction_ref_count_(0), tls_(false) {
    }
 
    virtual ~MySqlConnection();
 
    static std::vector<std::string>
    toKeaAdminParameters(ParameterMap const& params);
 
    static std::pair<uint32_t, uint32_t>
    getVersion(const ParameterMap& parameters,
               const IOServiceAccessorPtr& ac = IOServiceAccessorPtr(),
               const DbCallback& cb = DbCallback(),
               const std::string& timer_name = std::string());
 
    static void
    ensureSchemaVersion(const ParameterMap& parameters,
                        const DbCallback& cb = DbCallback(),
                        const std::string& timer_name = std::string());
 
    static void
    initializeSchema(const ParameterMap& parameters);
 
    void prepareStatement(uint32_t index, const char* text);
 
    void prepareStatements(const TaggedStatement* start_statement,
                           const TaggedStatement* end_statement);
 
    template<typename StatementIndex>
    MYSQL_STMT* getStatement(StatementIndex index) const {
        if (statements_[index]->mysql == 0) {
            isc_throw(db::DbConnectionUnusable,
                      "MySQL pointer for the prepared statement is NULL as a result of connectivity loss");
        }
        return (statements_[index]);
    }
 
    void openDatabase();
 
 
    static
    void convertToDatabaseTime(const time_t input_time, MYSQL_TIME& output_time);
 
    static
    void convertToDatabaseTime(const time_t cltt, const uint32_t valid_lifetime,
                               MYSQL_TIME& expire);
 
    static
    void convertFromDatabaseTime(const MYSQL_TIME& expire,
                                 uint32_t valid_lifetime, time_t& cltt);
 
    void startTransaction();
 
    bool isTransactionStarted() const;
 
    template<typename StatementIndex>
    void selectQuery(const StatementIndex& index,
                     const MySqlBindingCollection& in_bindings,
                     MySqlBindingCollection& out_bindings,
                     ConsumeResultFun process_result) {
        checkUnusable();
        // Extract native input bindings.
        std::vector<MYSQL_BIND> in_bind_vec;
        for (const MySqlBindingPtr& in_binding : in_bindings) {
            in_bind_vec.push_back(in_binding->getMySqlBinding());
        }
 
        int status = 0;
        if (!in_bind_vec.empty()) {
            // Bind parameters to the prepared statement.
            status = mysql_stmt_bind_param(getStatement(index),
                                           in_bind_vec.empty() ? 0 : &in_bind_vec[0]);
            checkError(status, index, "unable to bind parameters for select");
        }
 
        // Bind variables that will receive results as well.
        std::vector<MYSQL_BIND> out_bind_vec;
        for (const MySqlBindingPtr& out_binding : out_bindings) {
            out_bind_vec.push_back(out_binding->getMySqlBinding());
        }
        if (!out_bind_vec.empty()) {
            status = mysql_stmt_bind_result(getStatement(index), &out_bind_vec[0]);
            checkError(status, index, "unable to bind result parameters for select");
        }
 
        // Execute query.
        status = MysqlExecuteStatement(getStatement(index));
        checkError(status, index, "unable to execute");
 
        status = mysql_stmt_store_result(getStatement(index));
        checkError(status, index, "unable to set up for storing all results");
 
        // Fetch results.
        MySqlFreeResult fetch_release(getStatement(index));
        while ((status = mysql_stmt_fetch(getStatement(index))) ==
               MLM_MYSQL_FETCH_SUCCESS) {
            try {
                // For each returned row call user function which should
                // consume the row and copy the data to a safe place.
                process_result(out_bindings);
 
            } catch (const std::exception& ex) {
                // Rethrow the exception with a bit more data.
                isc_throw(BadValue, ex.what() << ". Statement is <" <<
                          text_statements_[index] << ">");
            }
        }
 
        // How did the fetch end?
        // If mysql_stmt_fetch return value is equal to 1 an error occurred.
        if (status == MLM_MYSQL_FETCH_FAILURE) {
            // Error - unable to fetch results
            checkError(status, index, "unable to fetch results");
 
        } else if (status == MYSQL_DATA_TRUNCATED) {
            // Data truncated - throw an exception indicating what was at fault
            isc_throw(DataTruncated, text_statements_[index]
                      << " returned truncated data");
        }
    }
 
    template<typename StatementIndex>
    void insertQuery(const StatementIndex& index,
                     const MySqlBindingCollection& in_bindings) {
        checkUnusable();
        std::vector<MYSQL_BIND> in_bind_vec;
        for (const MySqlBindingPtr& in_binding : in_bindings) {
            in_bind_vec.push_back(in_binding->getMySqlBinding());
        }
 
        // Bind the parameters to the statement
        int status = mysql_stmt_bind_param(getStatement(index),
                                           in_bind_vec.empty() ? 0 : &in_bind_vec[0]);
        checkError(status, index, "unable to bind parameters");
 
        // Execute the statement
        status = MysqlExecuteStatement(getStatement(index));
 
        if (status != 0) {
            // Failure: check for the special case of duplicate entry.
            if (mysql_errno(mysql_) == ER_DUP_ENTRY) {
                isc_throw(DuplicateEntry, "Database duplicate entry error");
            }
            // Failure: check for the special case of WHERE returning NULL.
            if (mysql_errno(mysql_) == ER_BAD_NULL_ERROR) {
                isc_throw(NullKeyError, "Database bad NULL error");
            }
            checkError(status, index, "unable to execute");
        }
    }
 
    template<typename StatementIndex>
    uint64_t updateDeleteQuery(const StatementIndex& index,
                               const MySqlBindingCollection& in_bindings) {
        checkUnusable();
        std::vector<MYSQL_BIND> in_bind_vec;
        for (const MySqlBindingPtr& in_binding : in_bindings) {
            in_bind_vec.push_back(in_binding->getMySqlBinding());
        }
 
        // Bind the parameters to the statement
        int status = mysql_stmt_bind_param(getStatement(index),
                                           in_bind_vec.empty() ? 0 : &in_bind_vec[0]);
        checkError(status, index, "unable to bind parameters");
 
        // Execute the statement
        status = MysqlExecuteStatement(getStatement(index));
 
        if (status != 0) {
            // Failure: check for the special case of duplicate entry.
            if ((mysql_errno(mysql_) == ER_DUP_ENTRY)
#ifdef ER_FOREIGN_DUPLICATE_KEY
                || (mysql_errno(mysql_) == ER_FOREIGN_DUPLICATE_KEY)
#endif
#ifdef ER_FOREIGN_DUPLICATE_KEY_WITH_CHILD_INFO
                || (mysql_errno(mysql_) == ER_FOREIGN_DUPLICATE_KEY_WITH_CHILD_INFO)
#endif
#ifdef ER_FOREIGN_DUPLICATE_KEY_WITHOUT_CHILD_INFO
                || (mysql_errno(mysql_) == ER_FOREIGN_DUPLICATE_KEY_WITHOUT_CHILD_INFO)
#endif
                ) {
                isc_throw(DuplicateEntry, "Database duplicate entry error");
            }
            checkError(status, index, "unable to execute");
        }
 
        // Let's return how many rows were affected.
        return (static_cast<uint64_t>(mysql_stmt_affected_rows(getStatement(index))));
    }
 
    void commit();
 
    void rollback();
 
    template<typename StatementIndex>
    void checkError(const int status, const StatementIndex& index,
                    const char* what) {
        if (status != 0) {
            switch(mysql_errno(mysql_)) {
                // These are the ones we consider fatal. Remember this method is
                // used to check errors of API calls made subsequent to successfully
                // connecting.  Errors occurring while attempting to connect are
                // checked in the connection code. An alternative would be to call
                // mysql_ping() - assuming autoreconnect is off. If that fails
                // then we know connection is toast.
            case CR_SERVER_GONE_ERROR:
            case CR_SERVER_LOST:
            case CR_OUT_OF_MEMORY:
            case CR_CONNECTION_ERROR: {
                DB_LOG_ERROR(db::MYSQL_FATAL_ERROR)
                    .arg(what)
                    .arg(text_statements_[static_cast<int>(index)])
                    .arg(mysql_error(mysql_))
                    .arg(mysql_errno(mysql_));
 
                // Mark this connection as no longer usable.
                markUnusable();
 
                // Start the connection recovery.
                startRecoverDbConnection();
 
                // We still need to throw so caller can error out of the current
                // processing.
                isc_throw(db::DbConnectionUnusable,
                          "fatal database error or connectivity lost");
            }
            default:
                // Connection is ok, so it must be an SQL error
                isc_throw(db::DbOperationError, what << " for <"
                          << text_statements_[static_cast<int>(index)]
                          << ">, reason: "
                          << mysql_error(mysql_) << " (error code "
                          << mysql_errno(mysql_) << ")");
            }
        }
    }
 
    void startRecoverDbConnection() {
        if (callback_) {
            if (!io_service_ && io_service_accessor_) {
                io_service_ = (*io_service_accessor_)();
                io_service_accessor_.reset();
            }
 
            if (io_service_) {
                io_service_->post(std::bind(callback_, reconnectCtl()));
            }
        }
    }
 
    bool getTls() const {
        return (tls_);
    }
 
    std::string getTlsCipher() {
        const char* cipher = mysql_get_ssl_cipher(mysql_);
        return (cipher ? std::string(cipher) : "");
    }
 
private:
 
    template<typename T>
    void setIntParameterValue(const std::string& name, int64_t min, int64_t max, T& value);
 
    std::vector<MYSQL_STMT*> statements_;
 
public:
 
    std::vector<std::string> text_statements_;
 
    MySqlHolder mysql_;
 
    IOServiceAccessorPtr io_service_accessor_;
 
    isc::asiolink::IOServicePtr io_service_;
 
    int transaction_ref_count_;
 
    bool tls_;
 
    static std::string KEA_ADMIN_;
};
 
} // end of isc::db namespace
} // end of isc namespace
 
#endif // MYSQL_CONNECTION_H