mirror of https://github.com/sysown/proxysql
This commit lays the groundwork for handling PostgreSQL’s extended query protocol (prepared statements) by introducing parsing, it's caching and execution framework stubs. - Implement Parse handling - Implement caching of parse (reuse) - Introduce Execute framework stub for future binding and execution logic.pull/5044/head
Rahim Kanji
10 months ago
parent
026f458f7c
commit
24fecc1f6e
@ -0,0 +1,234 @@
|
||||
#ifndef CLASS_PGSQL_PREPARED_STATEMENT_H
|
||||
#define CLASS_PGSQL_PREPARED_STATEMENT_H
|
||||
|
||||
#include "proxysql.h"
|
||||
#include "cpp.h"
|
||||
|
||||
/*
|
||||
One of the main challenge in handling prepared statement (PS) is that a single
|
||||
PS could be executed on multiple backends, and on each backend it could have a
|
||||
different stmt_id.
|
||||
For this reason ProxySQL returns to the client a stmt_id generated by the proxy
|
||||
itself, and internally maps client's stmt_id with the backend stmt_id.
|
||||
|
||||
The implementation in ProxySQL is, simplified, the follow:
|
||||
* when a client sends a MYSQL_COM_STMT_PREPARE, ProxySQL executes it to one of
|
||||
the backend
|
||||
* the backend returns a stmt_id. This stmt_id is NOT returned to the client. The
|
||||
stmt_id returned from the backend is stored in MySQL_STMTs_local(), and
|
||||
MySQL_STMTs_local() is responsible for mapping the connection's MYSQL_STMT
|
||||
and a global_stmt_id
|
||||
* the global_stmt_id is the stmt_id returned to the client
|
||||
* the global_stmt_id is used to locate the relevant MySQL_STMT_Global_info() in
|
||||
MySQL_STMT_Manager()
|
||||
* MySQL_STMT_Global_info() stores all metadata associated with a PS
|
||||
* MySQL_STMT_Manager() is responsible for storing all MySQL_STMT_Global_info()
|
||||
in global structures accessible and shareble by all threads.
|
||||
|
||||
To summarie the most important classes:
|
||||
* MySQL_STMT_Global_info() stores all metadata associated with a PS
|
||||
* MySQL_STMT_Manager() stores all the MySQL_STMT_Global_info(), indexes using
|
||||
a global_stmt_id that iis the stmt_id generated by ProxySQL and returned to
|
||||
the client
|
||||
* MySQL_STMTs_local() associate PS located in a backend connection to a
|
||||
global_stmt_id
|
||||
*/
|
||||
|
||||
// class MySQL_STMT_Global_info represents information about a MySQL Prepared Statement
|
||||
// it is an internal representation of prepared statement
|
||||
// it include all metadata associated with it
|
||||
|
||||
class PgSQL_STMT_Global_info {
|
||||
private:
|
||||
void compute_hash();
|
||||
public:
|
||||
pthread_rwlock_t rwlock_;
|
||||
uint64_t digest;
|
||||
PGSQL_QUERY_command PgQueryCmd;
|
||||
char * digest_text;
|
||||
uint64_t hash;
|
||||
char *username;
|
||||
char *schemaname;
|
||||
char *query;
|
||||
unsigned int query_length;
|
||||
int ref_count_client;
|
||||
int ref_count_server;
|
||||
uint64_t statement_id;
|
||||
char* first_comment;
|
||||
uint64_t total_mem_usage;
|
||||
struct describe {
|
||||
uint16_t num_fields;
|
||||
uint16_t num_params;
|
||||
uint8_t flags;
|
||||
} describe;
|
||||
|
||||
bool is_select_NOT_for_update;
|
||||
PgSQL_STMT_Global_info(uint64_t id, char *u, char *s, char *q, unsigned int ql, char *fc, uint64_t _h);
|
||||
void update_metadata(MYSQL_STMT *stmt);
|
||||
~PgSQL_STMT_Global_info();
|
||||
void calculate_mem_usage();
|
||||
};
|
||||
|
||||
#if 0
|
||||
// stmt_execute_metadata_t represent metadata required to run STMT_EXECUTE
|
||||
class pgsql_stmt_execute_metadata_t {
|
||||
public:
|
||||
uint32_t size;
|
||||
uint32_t stmt_id;
|
||||
uint8_t flags;
|
||||
uint16_t num_params;
|
||||
MYSQL_BIND *binds;
|
||||
my_bool *is_nulls;
|
||||
unsigned long *lengths;
|
||||
void *pkt;
|
||||
pgsql_stmt_execute_metadata_t() {
|
||||
size = 0;
|
||||
stmt_id = 0;
|
||||
binds=NULL;
|
||||
is_nulls=NULL;
|
||||
lengths=NULL;
|
||||
pkt=NULL;
|
||||
}
|
||||
~pgsql_stmt_execute_metadata_t() {
|
||||
if (binds)
|
||||
free(binds);
|
||||
binds = NULL;
|
||||
if (is_nulls)
|
||||
free(is_nulls);
|
||||
is_nulls = NULL;
|
||||
if (lengths)
|
||||
free(lengths);
|
||||
lengths = NULL;
|
||||
size = 0;
|
||||
stmt_id = 0;
|
||||
if (pkt) {
|
||||
free(pkt);
|
||||
pkt = NULL;
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// server side, metadata related to STMT_EXECUTE are stored in MYSQL_STMT itself
|
||||
// client side, they are stored in stmt_execute_metadata_t
|
||||
// MySQL_STMTs_meta maps stmt_execute_metadata_t with stmt_id
|
||||
class PgSQL_STMTs_meta {
|
||||
private:
|
||||
unsigned int num_entries;
|
||||
std::map<uint32_t, pgsql_stmt_execute_metadata_t*> m;
|
||||
public:
|
||||
PgSQL_STMTs_meta() {
|
||||
num_entries=0;
|
||||
}
|
||||
~PgSQL_STMTs_meta() {
|
||||
for (std::map<uint32_t, pgsql_stmt_execute_metadata_t*>::iterator it=m.begin(); it!=m.end(); ++it) {
|
||||
pgsql_stmt_execute_metadata_t*sem=it->second;
|
||||
delete sem;
|
||||
}
|
||||
}
|
||||
// we declare it here to be inline
|
||||
void insert(uint32_t global_statement_id, pgsql_stmt_execute_metadata_t*stmt_meta) {
|
||||
std::pair<std::map<uint32_t, pgsql_stmt_execute_metadata_t*>::iterator,bool> ret;
|
||||
ret=m.insert(std::make_pair(global_statement_id, stmt_meta));
|
||||
if (ret.second==true) {
|
||||
num_entries++;
|
||||
}
|
||||
}
|
||||
// we declare it here to be inline
|
||||
pgsql_stmt_execute_metadata_t* find(uint32_t global_statement_id) {
|
||||
auto s=m.find(global_statement_id);
|
||||
if (s!=m.end()) { // found
|
||||
return s->second;
|
||||
}
|
||||
return NULL; // not found
|
||||
}
|
||||
|
||||
void erase(uint32_t global_statement_id) {
|
||||
auto s=m.find(global_statement_id);
|
||||
if (s!=m.end()) { // found
|
||||
pgsql_stmt_execute_metadata_t*sem=s->second;
|
||||
delete sem;
|
||||
num_entries--;
|
||||
m.erase(s);
|
||||
}
|
||||
}
|
||||
};
|
||||
#endif
|
||||
|
||||
// class MySQL_STMTs_local associates a global statement ID with a local statement ID for a specific connection
|
||||
|
||||
class PgSQL_STMTs_local_v14 {
|
||||
private:
|
||||
bool is_client_;
|
||||
std::stack<uint32_t> free_backend_ids;
|
||||
uint32_t local_max_stmt_id = 0;
|
||||
|
||||
public:
|
||||
// this map associate client_stmt_id to global_stmt_id : this is used only for client connections
|
||||
std::map<std::string, uint64_t> stmt_name_to_global_ids;
|
||||
// this multimap associate global_stmt_id to client_stmt_id : this is used only for client connections
|
||||
std::multimap<uint64_t, std::string> global_id_to_stmt_names;
|
||||
|
||||
// this map associate backend_stmt_id to global_stmt_id : this is used only for backend connections
|
||||
std::map<uint32_t, uint64_t> backend_stmt_to_global_ids;
|
||||
// this map associate global_stmt_id to backend_stmt_id : this is used only for backend connections
|
||||
std::map<uint64_t, uint32_t> global_stmt_to_backend_ids;
|
||||
|
||||
PgSQL_Session *sess;
|
||||
PgSQL_STMTs_local_v14(bool _ic) : is_client_(_ic), sess(NULL) { }
|
||||
~PgSQL_STMTs_local_v14();
|
||||
|
||||
inline
|
||||
void set_is_client(PgSQL_Session *_s) {
|
||||
sess=_s;
|
||||
is_client_ = true;
|
||||
}
|
||||
|
||||
inline
|
||||
bool is_client() { return is_client_; }
|
||||
inline
|
||||
unsigned int get_num_backend_stmts() { return backend_stmt_to_global_ids.size(); }
|
||||
|
||||
void backend_insert(uint64_t global_stmt_id, uint32_t backend_stmt_id);
|
||||
void client_insert(uint64_t global_stmt_id, const std::string& client_stmt_name);
|
||||
uint64_t compute_hash(char *user, char *schema, char *query, unsigned int query_length);
|
||||
uint32_t generate_new_backend_stmt_id();
|
||||
uint64_t find_global_id_from_stmt_name(const std::string& client_stmt_name);
|
||||
bool client_close(const std::string& stmt_name);
|
||||
};
|
||||
|
||||
|
||||
class PgSQL_STMT_Manager_v14 {
|
||||
private:
|
||||
uint64_t next_statement_id;
|
||||
uint64_t num_stmt_with_ref_client_count_zero;
|
||||
uint64_t num_stmt_with_ref_server_count_zero;
|
||||
pthread_rwlock_t rwlock_;
|
||||
std::map<uint64_t, PgSQL_STMT_Global_info *> map_stmt_id_to_info; // map using statement id
|
||||
std::map<uint64_t, PgSQL_STMT_Global_info *> map_stmt_hash_to_info; // map using hashes
|
||||
std::stack<uint64_t> free_stmt_ids;
|
||||
struct {
|
||||
uint64_t c_unique;
|
||||
uint64_t c_total;
|
||||
uint64_t stmt_max_stmt_id;
|
||||
uint64_t cached;
|
||||
uint64_t s_unique;
|
||||
uint64_t s_total;
|
||||
} statuses;
|
||||
time_t last_purge_time;
|
||||
public:
|
||||
PgSQL_STMT_Manager_v14();
|
||||
~PgSQL_STMT_Manager_v14();
|
||||
PgSQL_STMT_Global_info * find_prepared_statement_by_hash(uint64_t hash);
|
||||
PgSQL_STMT_Global_info * find_prepared_statement_by_stmt_id(uint64_t id, bool lock=true);
|
||||
void rdlock() { pthread_rwlock_rdlock(&rwlock_); }
|
||||
void wrlock() { pthread_rwlock_wrlock(&rwlock_); }
|
||||
void unlock() { pthread_rwlock_unlock(&rwlock_); }
|
||||
void ref_count_client(uint64_t _stmt, int _v, bool lock=true);
|
||||
void ref_count_server(uint64_t _stmt, int _v, bool lock=true);
|
||||
PgSQL_STMT_Global_info * add_prepared_statement(char *u, char *s, char *q, unsigned int ql, char *fc, bool lock=true);
|
||||
void get_metrics(uint64_t *c_unique, uint64_t *c_total, uint64_t *stmt_max_stmt_id, uint64_t *cached, uint64_t *s_unique, uint64_t *s_total);
|
||||
SQLite3_result* get_prepared_statements_global_infos();
|
||||
void get_memory_usage(uint64_t& prep_stmt_metadata_mem_usage, uint64_t& prep_stmt_backend_mem_usage);
|
||||
};
|
||||
|
||||
#endif /* CLASS_PGSQL_PREPARED_STATEMENT_H */
|
||||
@ -0,0 +1,908 @@
|
||||
#include "proxysql.h"
|
||||
#include "cpp.h"
|
||||
|
||||
#ifndef SPOOKYV2
|
||||
#include "SpookyV2.h"
|
||||
#define SPOOKYV2
|
||||
#endif
|
||||
|
||||
#include "PgSQL_PreparedStatement.h"
|
||||
#include "MySQL_Protocol.h"
|
||||
|
||||
//extern MySQL_STMT_Manager *GloMyStmt;
|
||||
//static uint32_t add_prepared_statement_calls = 0;
|
||||
//static uint32_t find_prepared_statement_by_hash_calls = 0;
|
||||
//#else
|
||||
extern PgSQL_STMT_Manager_v14 *GloPgStmt;
|
||||
//#endif
|
||||
|
||||
const int PS_GLOBAL_STATUS_FIELD_NUM = 9;
|
||||
|
||||
static uint64_t stmt_compute_hash(char *user,
|
||||
char *schema, char *query,
|
||||
unsigned int query_length) {
|
||||
int l = 0;
|
||||
l += strlen(user);
|
||||
l += strlen(schema);
|
||||
// two random seperators
|
||||
#define _COMPUTE_HASH_DEL1_ "-ujhtgf76y576574fhYTRDFwdt-"
|
||||
#define _COMPUTE_HASH_DEL2_ "-8k7jrhtrgJHRgrefgreRFewg6-"
|
||||
l += strlen(_COMPUTE_HASH_DEL1_);
|
||||
l += strlen(_COMPUTE_HASH_DEL2_);
|
||||
l += query_length;
|
||||
char *buf = (char *)malloc(l);
|
||||
l = 0;
|
||||
|
||||
// write user
|
||||
strcpy(buf + l, user);
|
||||
l += strlen(user);
|
||||
|
||||
// write delimiter1
|
||||
strcpy(buf + l, _COMPUTE_HASH_DEL1_);
|
||||
l += strlen(_COMPUTE_HASH_DEL1_);
|
||||
|
||||
// write schema
|
||||
strcpy(buf + l, schema);
|
||||
l += strlen(schema);
|
||||
|
||||
// write delimiter2
|
||||
strcpy(buf + l, _COMPUTE_HASH_DEL2_);
|
||||
l += strlen(_COMPUTE_HASH_DEL2_);
|
||||
|
||||
// write query
|
||||
memcpy(buf + l, query, query_length);
|
||||
l += query_length;
|
||||
|
||||
uint64_t hash = SpookyHash::Hash64(buf, l, 0);
|
||||
free(buf);
|
||||
return hash;
|
||||
}
|
||||
|
||||
void PgSQL_STMT_Global_info::compute_hash() {
|
||||
hash = stmt_compute_hash(username, schemaname, query,
|
||||
query_length);
|
||||
}
|
||||
|
||||
PgSQL_STMT_Global_info::PgSQL_STMT_Global_info(uint64_t id,
|
||||
char *u, char *s, char *q,
|
||||
unsigned int ql,
|
||||
char *fc,
|
||||
uint64_t _h) {
|
||||
pthread_rwlock_init(&rwlock_, NULL);
|
||||
total_mem_usage = 0;
|
||||
statement_id = id;
|
||||
ref_count_client = 0;
|
||||
ref_count_server = 0;
|
||||
digest_text = NULL;
|
||||
username = strdup(u);
|
||||
schemaname = strdup(s);
|
||||
query = (char *)malloc(ql + 1);
|
||||
memcpy(query, q, ql);
|
||||
query[ql] = '\0'; // add NULL byte
|
||||
query_length = ql;
|
||||
if (fc) {
|
||||
first_comment = strdup(fc);
|
||||
} else {
|
||||
first_comment = NULL;
|
||||
}
|
||||
PgQueryCmd = PGSQL_QUERY__UNINITIALIZED;
|
||||
|
||||
if (_h) {
|
||||
hash = _h;
|
||||
} else {
|
||||
compute_hash();
|
||||
}
|
||||
|
||||
is_select_NOT_for_update = false;
|
||||
{ // see bug #899 . Most of the code is borrowed from
|
||||
// Query_Info::is_select_NOT_for_update()
|
||||
if (ql >= 7) {
|
||||
if (strncasecmp(q, (char *)"SELECT ", 7) == 0) { // is a SELECT
|
||||
if (ql >= 17) {
|
||||
char *p = q;
|
||||
p += ql - 11;
|
||||
if (strncasecmp(p, " FOR UPDATE", 11) == 0) { // is a SELECT FOR UPDATE
|
||||
__sync_fetch_and_add(&MyHGM->status.select_for_update_or_equivalent, 1);
|
||||
goto __exit_PgSQL_STMT_Global_info___search_select;
|
||||
}
|
||||
p = q;
|
||||
p += ql-10;
|
||||
if (strncasecmp(p, " FOR SHARE", 10) == 0) { // is a SELECT FOR SHARE
|
||||
__sync_fetch_and_add(&MyHGM->status.select_for_update_or_equivalent, 1);
|
||||
goto __exit_PgSQL_STMT_Global_info___search_select;
|
||||
}
|
||||
if (ql >= 25) {
|
||||
p = q;
|
||||
p += ql-19;
|
||||
if (strncasecmp(p, " LOCK IN SHARE MODE", 19) == 0) { // is a SELECT LOCK IN SHARE MODE
|
||||
__sync_fetch_and_add(&MyHGM->status.select_for_update_or_equivalent, 1);
|
||||
goto __exit_PgSQL_STMT_Global_info___search_select;
|
||||
}
|
||||
p = q;
|
||||
p += ql-7;
|
||||
if (strncasecmp(p," NOWAIT",7)==0) {
|
||||
// let simplify. If NOWAIT is used, we assume FOR UPDATE|SHARE is used
|
||||
__sync_fetch_and_add(&MyHGM->status.select_for_update_or_equivalent, 1);
|
||||
goto __exit_PgSQL_STMT_Global_info___search_select;
|
||||
/*
|
||||
if (strcasestr(q," FOR UPDATE ")) {
|
||||
__sync_fetch_and_add(&MyHGM->status.select_for_update_or_equivalent, 1);
|
||||
goto __exit_PgSQL_STMT_Global_info___search_select;
|
||||
}
|
||||
if (strcasestr(q," FOR SHARE ")) {
|
||||
__sync_fetch_and_add(&MyHGM->status.select_for_update_or_equivalent, 1);
|
||||
goto __exit_PgSQL_STMT_Global_info___search_select;
|
||||
}
|
||||
*/
|
||||
}
|
||||
p = q;
|
||||
p += ql-12;
|
||||
if (strncasecmp(p," SKIP LOCKED",12)==0) {
|
||||
// let simplify. If SKIP LOCKED is used, we assume FOR UPDATE|SHARE is used
|
||||
__sync_fetch_and_add(&MyHGM->status.select_for_update_or_equivalent, 1);
|
||||
goto __exit_PgSQL_STMT_Global_info___search_select;
|
||||
/*
|
||||
if (strcasestr(q," FOR UPDATE ")==NULL) {
|
||||
__sync_fetch_and_add(&MyHGM->status.select_for_update_or_equivalent, 1);
|
||||
goto __exit_PgSQL_STMT_Global_info___search_select;
|
||||
}
|
||||
if (strcasestr(q," FOR SHARE ")==NULL) {
|
||||
__sync_fetch_and_add(&MyHGM->status.select_for_update_or_equivalent, 1);
|
||||
goto __exit_PgSQL_STMT_Global_info___search_select;
|
||||
}
|
||||
*/
|
||||
}
|
||||
p=q;
|
||||
char buf[129];
|
||||
if (ql>=128) { // for long query, just check the last 128 bytes
|
||||
p+=ql-128;
|
||||
memcpy(buf,p,128);
|
||||
buf[128]=0;
|
||||
} else {
|
||||
memcpy(buf,p,ql);
|
||||
buf[ql]=0;
|
||||
}
|
||||
if (strcasestr(buf," FOR ")) {
|
||||
if (strcasestr(buf," FOR UPDATE ")) {
|
||||
__sync_fetch_and_add(&MyHGM->status.select_for_update_or_equivalent, 1);
|
||||
goto __exit_PgSQL_STMT_Global_info___search_select;
|
||||
}
|
||||
if (strcasestr(buf," FOR SHARE ")) {
|
||||
__sync_fetch_and_add(&MyHGM->status.select_for_update_or_equivalent, 1);
|
||||
goto __exit_PgSQL_STMT_Global_info___search_select;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
is_select_NOT_for_update = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
__exit_PgSQL_STMT_Global_info___search_select:
|
||||
calculate_mem_usage();
|
||||
}
|
||||
|
||||
void PgSQL_STMT_Global_info::calculate_mem_usage() {
|
||||
total_mem_usage = sizeof(PgSQL_STMT_Global_info) +
|
||||
query_length + 1;// +
|
||||
//(ref_count_client * 24) +
|
||||
//(ref_count_server * 24);
|
||||
|
||||
if (username) total_mem_usage += strlen(username) + 1;
|
||||
if (schemaname) total_mem_usage += strlen(schemaname) + 1;
|
||||
if (first_comment) total_mem_usage += strlen(first_comment) + 1;
|
||||
if (digest_text) total_mem_usage += strlen(digest_text) + 1;
|
||||
}
|
||||
|
||||
void PgSQL_STMT_Global_info::update_metadata(MYSQL_STMT *stmt) {
|
||||
|
||||
bool need_refresh = false;
|
||||
pthread_rwlock_wrlock(&rwlock_);
|
||||
/* if (
|
||||
(num_params != stmt->param_count)
|
||||
||
|
||||
(num_columns != stmt->field_count)
|
||||
) {
|
||||
need_refresh = true;
|
||||
}
|
||||
for (i = 0; i < num_columns; i++) {
|
||||
if (need_refresh == false) { // don't bother to check if need_refresh == true
|
||||
bool ok = true;
|
||||
MYSQL_FIELD *fs = &(stmt->fields[i]);
|
||||
MYSQL_FIELD *fd = fields[i];
|
||||
if (ok) {
|
||||
ok = false;
|
||||
if (fd->name == NULL && fs->name == NULL) {
|
||||
ok = true;
|
||||
} else {
|
||||
if (fd->name && fs->name && strcmp(fd->name,fs->name)==0) {
|
||||
ok = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (ok) {
|
||||
ok = false;
|
||||
if (fd->org_name == NULL && fs->org_name == NULL) {
|
||||
ok = true;
|
||||
} else {
|
||||
if (fd->org_name && fs->org_name && strcmp(fd->org_name,fs->org_name)==0) {
|
||||
ok = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (ok) {
|
||||
ok = false;
|
||||
if (fd->table == NULL && fs->table == NULL) {
|
||||
ok = true;
|
||||
} else {
|
||||
if (fd->table && fs->table && strcmp(fd->table,fs->table)==0) {
|
||||
ok = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (ok) {
|
||||
ok = false;
|
||||
if (fd->org_table == NULL && fs->org_table == NULL) {
|
||||
ok = true;
|
||||
} else {
|
||||
if (fd->org_table && fs->org_table && strcmp(fd->org_table,fs->org_table)==0) {
|
||||
ok = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (ok) {
|
||||
ok = false;
|
||||
if (fd->db == NULL && fs->db == NULL) {
|
||||
ok = true;
|
||||
} else {
|
||||
if (fd->db && fs->db && strcmp(fd->db,fs->db)==0) {
|
||||
ok = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (ok) {
|
||||
ok = false;
|
||||
if (fd->catalog == NULL && fs->catalog == NULL) {
|
||||
ok = true;
|
||||
} else {
|
||||
if (fd->catalog && fs->catalog && strcmp(fd->catalog,fs->catalog)==0) {
|
||||
ok = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (ok) {
|
||||
ok = false;
|
||||
if (fd->def == NULL && fs->def == NULL) {
|
||||
ok = true;
|
||||
} else {
|
||||
if (fd->def && fs->def && strcmp(fd->def,fs->def)==0) {
|
||||
ok = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (ok == false) {
|
||||
need_refresh = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (need_refresh) {
|
||||
if (digest_text && strncasecmp(digest_text, "EXPLAIN", strlen("EXPLAIN"))==0) {
|
||||
// do not print any message in case of EXPLAIN
|
||||
} else {
|
||||
proxy_warning("Updating metadata for stmt %lu , user %s, query %s\n", statement_id, username, query);
|
||||
}
|
||||
// from here is copied from destructor
|
||||
if (num_columns) {
|
||||
uint16_t i;
|
||||
for (i = 0; i < num_columns; i++) {
|
||||
MYSQL_FIELD *f = fields[i];
|
||||
if (f->name) {
|
||||
free(f->name);
|
||||
f->name = NULL;
|
||||
}
|
||||
if (f->org_name) {
|
||||
free(f->org_name);
|
||||
f->org_name = NULL;
|
||||
}
|
||||
if (f->table) {
|
||||
free(f->table);
|
||||
f->table = NULL;
|
||||
}
|
||||
if (f->org_table) {
|
||||
free(f->org_table);
|
||||
f->org_table = NULL;
|
||||
}
|
||||
if (f->db) {
|
||||
free(f->db);
|
||||
f->db = NULL;
|
||||
}
|
||||
if (f->catalog) {
|
||||
free(f->catalog);
|
||||
f->catalog = NULL;
|
||||
}
|
||||
if (f->def) {
|
||||
free(f->def);
|
||||
f->def = NULL;
|
||||
}
|
||||
free(fields[i]);
|
||||
}
|
||||
free(fields);
|
||||
fields = NULL;
|
||||
}
|
||||
if (num_params) {
|
||||
uint16_t i;
|
||||
for (i = 0; i < num_params; i++) {
|
||||
free(params[i]);
|
||||
}
|
||||
free(params);
|
||||
params = NULL;
|
||||
}
|
||||
// till here is copied from destructor
|
||||
|
||||
// from here is copied from constructor
|
||||
num_params = stmt->param_count;
|
||||
num_columns = stmt->field_count;
|
||||
fields = NULL;
|
||||
if (num_columns) {
|
||||
fields = (MYSQL_FIELD **)malloc(num_columns * sizeof(MYSQL_FIELD *));
|
||||
uint16_t i;
|
||||
for (i = 0; i < num_columns; i++) {
|
||||
fields[i] = (MYSQL_FIELD *)malloc(sizeof(MYSQL_FIELD));
|
||||
MYSQL_FIELD *fs = &(stmt->fields[i]);
|
||||
MYSQL_FIELD *fd = fields[i];
|
||||
// first copy all fields
|
||||
memcpy(fd, fs, sizeof(MYSQL_FIELD));
|
||||
// then duplicate strings
|
||||
fd->name = (fs->name ? strdup(fs->name) : NULL);
|
||||
fd->org_name = (fs->org_name ? strdup(fs->org_name) : NULL);
|
||||
fd->table = (fs->table ? strdup(fs->table) : NULL);
|
||||
fd->org_table = (fs->org_table ? strdup(fs->org_table) : NULL);
|
||||
fd->db = (fs->db ? strdup(fs->db) : NULL);
|
||||
fd->catalog = (fs->catalog ? strdup(fs->catalog) : NULL);
|
||||
fd->def = (fs->def ? strdup(fs->def) : NULL);
|
||||
}
|
||||
}
|
||||
|
||||
params = NULL;
|
||||
if (num_params == 2) {
|
||||
PROXY_TRACE();
|
||||
}
|
||||
if (num_params) {
|
||||
params = (MYSQL_BIND **)malloc(num_params * sizeof(MYSQL_BIND *));
|
||||
uint16_t i;
|
||||
for (i = 0; i < num_params; i++) {
|
||||
params[i] = (MYSQL_BIND *)malloc(sizeof(MYSQL_BIND));
|
||||
// MYSQL_BIND *ps=&(stmt->params[i]);
|
||||
// MYSQL_BIND *pd=params[i];
|
||||
// copy all params
|
||||
// memcpy(pd,ps,sizeof(MYSQL_BIND));
|
||||
memset(params[i], 0, sizeof(MYSQL_BIND));
|
||||
}
|
||||
}
|
||||
|
||||
// till here is copied from constructor
|
||||
calculate_mem_usage();
|
||||
}*/
|
||||
pthread_rwlock_unlock(&rwlock_);
|
||||
}
|
||||
|
||||
PgSQL_STMT_Global_info::~PgSQL_STMT_Global_info() {
|
||||
free(username);
|
||||
free(schemaname);
|
||||
free(query);
|
||||
if (first_comment) {
|
||||
free(first_comment);
|
||||
}
|
||||
/*if (num_columns) {
|
||||
uint16_t i;
|
||||
for (i = 0; i < num_columns; i++) {
|
||||
MYSQL_FIELD *f = fields[i];
|
||||
if (f->name) {
|
||||
free(f->name);
|
||||
f->name = NULL;
|
||||
}
|
||||
if (f->org_name) {
|
||||
free(f->org_name);
|
||||
f->org_name = NULL;
|
||||
}
|
||||
if (f->table) {
|
||||
free(f->table);
|
||||
f->table = NULL;
|
||||
}
|
||||
if (f->org_table) {
|
||||
free(f->org_table);
|
||||
f->org_table = NULL;
|
||||
}
|
||||
if (f->db) {
|
||||
free(f->db);
|
||||
f->db = NULL;
|
||||
}
|
||||
if (f->catalog) {
|
||||
free(f->catalog);
|
||||
f->catalog = NULL;
|
||||
}
|
||||
if (f->def) {
|
||||
free(f->def);
|
||||
f->def = NULL;
|
||||
}
|
||||
free(fields[i]);
|
||||
}
|
||||
free(fields);
|
||||
fields = NULL;
|
||||
}
|
||||
|
||||
if (num_params) {
|
||||
uint16_t i;
|
||||
for (i = 0; i < num_params; i++) {
|
||||
free(params[i]);
|
||||
}
|
||||
free(params);
|
||||
params = NULL;
|
||||
}
|
||||
*/
|
||||
if (digest_text) {
|
||||
free(digest_text);
|
||||
digest_text = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
void PgSQL_STMTs_local_v14::backend_insert(uint64_t global_stmt_id, uint32_t backend_stmt_id) {
|
||||
//std::pair<std::map<uint64_t, MYSQL_STMT *>::iterator, bool> ret;
|
||||
//ret = global_stmt_to_backend_stmt.insert(std::make_pair(global_statement_id, stmt));
|
||||
global_stmt_to_backend_ids.insert(std::make_pair(global_stmt_id, backend_stmt_id));
|
||||
backend_stmt_to_global_ids.insert(std::make_pair(backend_stmt_id,global_stmt_id));
|
||||
// note: backend_insert() is always called after add_prepared_statement()
|
||||
// for this reason, we will the ref count increase in add_prepared_statement()
|
||||
// GloPgStmt->ref_count_client(global_statement_id, 1);
|
||||
}
|
||||
|
||||
void PgSQL_STMTs_local_v14::client_insert(uint64_t global_stmt_id, const std::string& client_stmt_name) {
|
||||
stmt_name_to_global_ids.insert(std::make_pair(client_stmt_name, global_stmt_id));
|
||||
global_id_to_stmt_names.insert(std::make_pair(global_stmt_id, client_stmt_name));
|
||||
}
|
||||
|
||||
uint64_t PgSQL_STMTs_local_v14::compute_hash(char *user,
|
||||
char *schema, char *query,
|
||||
unsigned int query_length) {
|
||||
uint64_t hash;
|
||||
hash = stmt_compute_hash(user, schema, query, query_length);
|
||||
return hash;
|
||||
}
|
||||
|
||||
PgSQL_STMT_Manager_v14::PgSQL_STMT_Manager_v14() {
|
||||
last_purge_time = time(NULL);
|
||||
pthread_rwlock_init(&rwlock_, NULL);
|
||||
map_stmt_id_to_info= std::map<uint64_t, PgSQL_STMT_Global_info *>(); // map using statement id
|
||||
map_stmt_hash_to_info = std::map<uint64_t, PgSQL_STMT_Global_info *>(); // map using hashes
|
||||
free_stmt_ids = std::stack<uint64_t> ();
|
||||
|
||||
next_statement_id =
|
||||
1; // we initialize this as 1 because we 0 is not allowed
|
||||
num_stmt_with_ref_client_count_zero = 0;
|
||||
num_stmt_with_ref_server_count_zero = 0;
|
||||
statuses.c_unique = 0;
|
||||
statuses.c_total = 0;
|
||||
statuses.stmt_max_stmt_id = 0;
|
||||
statuses.cached = 0;
|
||||
statuses.s_unique = 0;
|
||||
statuses.s_total = 0;
|
||||
}
|
||||
|
||||
PgSQL_STMT_Manager_v14::~PgSQL_STMT_Manager_v14() {
|
||||
for (auto it = map_stmt_id_to_info.begin(); it != map_stmt_id_to_info.end(); ++it) {
|
||||
PgSQL_STMT_Global_info * a = it->second;
|
||||
delete a;
|
||||
}
|
||||
}
|
||||
|
||||
void PgSQL_STMT_Manager_v14::ref_count_client(uint64_t _stmt_id ,int _v, bool lock) {
|
||||
if (lock)
|
||||
pthread_rwlock_wrlock(&rwlock_);
|
||||
auto s = map_stmt_id_to_info.find(_stmt_id);
|
||||
if (s != map_stmt_id_to_info.end()) {
|
||||
statuses.c_total += _v;
|
||||
PgSQL_STMT_Global_info *stmt_info = s->second;
|
||||
if (stmt_info->ref_count_client == 0 && _v == 1) {
|
||||
__sync_sub_and_fetch(&num_stmt_with_ref_client_count_zero,1);
|
||||
} else {
|
||||
if (stmt_info->ref_count_client == 1 && _v == -1) {
|
||||
__sync_add_and_fetch(&num_stmt_with_ref_client_count_zero,1);
|
||||
}
|
||||
}
|
||||
stmt_info->ref_count_client += _v;
|
||||
time_t ct = time(NULL);
|
||||
uint64_t num_client_count_zero = __sync_add_and_fetch(&num_stmt_with_ref_client_count_zero, 0);
|
||||
uint64_t num_server_count_zero = __sync_add_and_fetch(&num_stmt_with_ref_server_count_zero, 0);
|
||||
|
||||
size_t map_size = map_stmt_id_to_info.size();
|
||||
if (
|
||||
(ct > last_purge_time+1) &&
|
||||
(map_size > (unsigned)mysql_thread___max_stmts_cache ) &&
|
||||
(num_client_count_zero > map_size/10) &&
|
||||
(num_server_count_zero > map_size/10)
|
||||
) { // purge only if there is at least 10% gain
|
||||
last_purge_time = ct;
|
||||
int max_purge = map_size ;
|
||||
int i = -1;
|
||||
uint64_t *torem =
|
||||
(uint64_t *)malloc(max_purge * sizeof(uint64_t));
|
||||
for (std::map<uint64_t, PgSQL_STMT_Global_info*>::iterator it =
|
||||
map_stmt_id_to_info.begin();
|
||||
it != map_stmt_id_to_info.end(); ++it) {
|
||||
if ( (i == (max_purge - 1)) || (i == ((int)num_client_count_zero - 1)) ) {
|
||||
break; // nothing left to clean up
|
||||
}
|
||||
PgSQL_STMT_Global_info *a = it->second;
|
||||
if ((__sync_add_and_fetch(&a->ref_count_client, 0) == 0) &&
|
||||
(a->ref_count_server == 0) ) // this to avoid that IDs are incorrectly reused
|
||||
{
|
||||
uint64_t hash = a->hash;
|
||||
auto s2 = map_stmt_hash_to_info.find(hash);
|
||||
if (s2 != map_stmt_hash_to_info.end()) {
|
||||
map_stmt_hash_to_info.erase(s2);
|
||||
}
|
||||
__sync_sub_and_fetch(&num_stmt_with_ref_client_count_zero,1);
|
||||
//if (a->ref_count_server == 0) {
|
||||
//__sync_sub_and_fetch(&num_stmt_with_ref_server_count_zero,1);
|
||||
//}
|
||||
// m.erase(it);
|
||||
// delete a;
|
||||
i++;
|
||||
torem[i] = it->first;
|
||||
}
|
||||
}
|
||||
while (i >= 0) {
|
||||
uint64_t id = torem[i];
|
||||
auto s3 = map_stmt_id_to_info.find(id);
|
||||
PgSQL_STMT_Global_info *a = s3->second;
|
||||
if (a->ref_count_server == 0) {
|
||||
__sync_sub_and_fetch(&num_stmt_with_ref_server_count_zero,1);
|
||||
free_stmt_ids.push(id);
|
||||
}
|
||||
map_stmt_id_to_info.erase(s3);
|
||||
statuses.s_total -= a->ref_count_server;
|
||||
delete a;
|
||||
i--;
|
||||
}
|
||||
free(torem);
|
||||
}
|
||||
}
|
||||
if (lock)
|
||||
pthread_rwlock_unlock(&rwlock_);
|
||||
}
|
||||
|
||||
void PgSQL_STMT_Manager_v14::ref_count_server(uint64_t _stmt_id ,int _v, bool lock) {
|
||||
if (lock)
|
||||
pthread_rwlock_wrlock(&rwlock_);
|
||||
std::map<uint64_t, PgSQL_STMT_Global_info *>::iterator s;
|
||||
s = map_stmt_id_to_info.find(_stmt_id);
|
||||
if (s != map_stmt_id_to_info.end()) {
|
||||
statuses.s_total += _v;
|
||||
PgSQL_STMT_Global_info *stmt_info = s->second;
|
||||
if (stmt_info->ref_count_server == 0 && _v == 1) {
|
||||
__sync_sub_and_fetch(&num_stmt_with_ref_server_count_zero,1);
|
||||
} else {
|
||||
if (stmt_info->ref_count_server == 1 && _v == -1) {
|
||||
__sync_add_and_fetch(&num_stmt_with_ref_server_count_zero,1);
|
||||
}
|
||||
}
|
||||
stmt_info->ref_count_server += _v;
|
||||
}
|
||||
if (lock)
|
||||
pthread_rwlock_unlock(&rwlock_);
|
||||
}
|
||||
|
||||
PgSQL_STMTs_local_v14::~PgSQL_STMTs_local_v14() {
|
||||
// Note: we do not free the prepared statements because we assume that
|
||||
// if we call this destructor the connection is being destroyed anyway
|
||||
|
||||
if (is_client_) {
|
||||
for (auto it = stmt_name_to_global_ids.begin();
|
||||
it != stmt_name_to_global_ids.end(); ++it) {
|
||||
uint64_t global_stmt_id = it->second;
|
||||
GloPgStmt->ref_count_client(global_stmt_id, -1);
|
||||
}
|
||||
} else {
|
||||
/*for (std::map<uint64_t, MYSQL_STMT*>::iterator it = global_stmt_to_backend_stmt.begin();
|
||||
it != global_stmt_to_backend_stmt.end(); ++it) {
|
||||
uint64_t global_stmt_id = it->first;
|
||||
MYSQL_STMT *stmt = it->second;
|
||||
proxy_mysql_stmt_close(stmt);
|
||||
GloPgStmt->ref_count_server(global_stmt_id, -1);
|
||||
}*/
|
||||
for (auto it = backend_stmt_to_global_ids.begin();
|
||||
it != backend_stmt_to_global_ids.end(); ++it) {
|
||||
uint64_t global_stmt_id = it->second;
|
||||
GloPgStmt->ref_count_server(global_stmt_id, -1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
PgSQL_STMT_Global_info *PgSQL_STMT_Manager_v14::find_prepared_statement_by_hash(uint64_t hash) {
|
||||
PgSQL_STMT_Global_info *ret = NULL; // assume we do not find it
|
||||
auto s = map_stmt_hash_to_info.find(hash);
|
||||
if (s != map_stmt_hash_to_info.end()) {
|
||||
ret = s->second;
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
PgSQL_STMT_Global_info* PgSQL_STMT_Manager_v14::find_prepared_statement_by_stmt_id(
|
||||
uint64_t id, bool lock) {
|
||||
PgSQL_STMT_Global_info*ret = NULL; // assume we do not find it
|
||||
if (lock) {
|
||||
pthread_rwlock_wrlock(&rwlock_);
|
||||
}
|
||||
|
||||
auto s = map_stmt_id_to_info.find(id);
|
||||
if (s != map_stmt_id_to_info.end()) {
|
||||
ret = s->second;
|
||||
}
|
||||
|
||||
if (lock) {
|
||||
pthread_rwlock_unlock(&rwlock_);
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
uint32_t PgSQL_STMTs_local_v14::generate_new_backend_stmt_id() {
|
||||
assert(is_client_ == false);
|
||||
if (free_backend_ids.empty() == false) {
|
||||
uint32_t backend_stmt_id = free_backend_ids.top();
|
||||
free_backend_ids.pop();
|
||||
return backend_stmt_id;
|
||||
}
|
||||
local_max_stmt_id++;
|
||||
return local_max_stmt_id;
|
||||
}
|
||||
|
||||
uint64_t PgSQL_STMTs_local_v14::find_global_id_from_stmt_name(const std::string& client_stmt_name) {
|
||||
uint64_t ret=0;
|
||||
auto s = stmt_name_to_global_ids.find(client_stmt_name);
|
||||
if (s != stmt_name_to_global_ids.end()) {
|
||||
ret = s->second;
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
bool PgSQL_STMTs_local_v14::client_close(const std::string& stmt_name) {
|
||||
auto s = stmt_name_to_global_ids.find(stmt_name);
|
||||
if (s != stmt_name_to_global_ids.end()) { // found
|
||||
uint64_t global_stmt_id = s->second;
|
||||
stmt_name_to_global_ids.erase(s);
|
||||
GloPgStmt->ref_count_client(global_stmt_id, -1);
|
||||
std::pair<std::multimap<uint64_t,std::string>::iterator, std::multimap<uint64_t,std::string>::iterator> ret;
|
||||
ret = global_id_to_stmt_names.equal_range(global_stmt_id);
|
||||
for (std::multimap<uint64_t, std::string>::iterator it=ret.first; it!=ret.second; ++it) {
|
||||
if (it->second == stmt_name) {
|
||||
global_id_to_stmt_names.erase(it);
|
||||
break;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
return false; // we don't really remove the prepared statement
|
||||
}
|
||||
|
||||
PgSQL_STMT_Global_info* PgSQL_STMT_Manager_v14::add_prepared_statement(
|
||||
char *u, char *s, char *q, unsigned int ql,
|
||||
char *fc, bool lock) {
|
||||
PgSQL_STMT_Global_info *ret = NULL;
|
||||
uint64_t hash = stmt_compute_hash(
|
||||
u, s, q, ql); // this identifies the prepared statement
|
||||
if (lock) {
|
||||
pthread_rwlock_wrlock(&rwlock_);
|
||||
}
|
||||
// try to find the statement
|
||||
auto f = map_stmt_hash_to_info.find(hash);
|
||||
if (f != map_stmt_hash_to_info.end()) {
|
||||
// found it!
|
||||
ret = f->second;
|
||||
ret->update_metadata(nullptr);
|
||||
} else {
|
||||
uint64_t next_id = 0;
|
||||
if (!free_stmt_ids.empty()) {
|
||||
next_id = free_stmt_ids.top();
|
||||
free_stmt_ids.pop();
|
||||
} else {
|
||||
next_id = next_statement_id;
|
||||
next_statement_id++;
|
||||
}
|
||||
|
||||
std::unique_ptr<PgSQL_STMT_Global_info> stmt_info (new PgSQL_STMT_Global_info(next_id, u, s, q, ql, fc, hash));
|
||||
// insert it in both maps
|
||||
map_stmt_id_to_info.insert(std::make_pair(stmt_info->statement_id, stmt_info.get()));
|
||||
map_stmt_hash_to_info.insert(std::make_pair(stmt_info->hash, stmt_info.get()));
|
||||
ret = stmt_info.release();
|
||||
__sync_add_and_fetch(&num_stmt_with_ref_client_count_zero,1);
|
||||
__sync_add_and_fetch(&num_stmt_with_ref_server_count_zero,1);
|
||||
}
|
||||
if (ret->ref_count_server == 0) {
|
||||
__sync_sub_and_fetch(&num_stmt_with_ref_server_count_zero,1);
|
||||
}
|
||||
ret->ref_count_server++;
|
||||
statuses.s_total++;
|
||||
if (lock) {
|
||||
pthread_rwlock_unlock(&rwlock_);
|
||||
}
|
||||
return ret;
|
||||
}
|
||||
|
||||
|
||||
void PgSQL_STMT_Manager_v14::get_memory_usage(uint64_t& prep_stmt_metadata_mem_usage, uint64_t& prep_stmt_backend_mem_usage) {
|
||||
prep_stmt_backend_mem_usage = 0;
|
||||
prep_stmt_metadata_mem_usage = sizeof(PgSQL_STMT_Manager_v14);
|
||||
rdlock();
|
||||
prep_stmt_metadata_mem_usage += map_stmt_id_to_info.size() * (sizeof(uint64_t) + sizeof(PgSQL_STMT_Global_info*));
|
||||
prep_stmt_metadata_mem_usage += map_stmt_hash_to_info.size() * (sizeof(uint64_t) + sizeof(PgSQL_STMT_Global_info*));
|
||||
prep_stmt_metadata_mem_usage += free_stmt_ids.size() * (sizeof(uint64_t));
|
||||
for (const auto& keyval : map_stmt_id_to_info) {
|
||||
const PgSQL_STMT_Global_info* stmt_global_info = keyval.second;
|
||||
prep_stmt_metadata_mem_usage += stmt_global_info->total_mem_usage;
|
||||
prep_stmt_metadata_mem_usage += stmt_global_info->ref_count_server;// *
|
||||
//((stmt_global_info->num_params * sizeof(MYSQL_BIND)) +
|
||||
//(stmt_global_info->num_columns * sizeof(MYSQL_FIELD))) + 16; // ~16 bytes of memory utilized by global_stmt_id and stmt_id mappings
|
||||
prep_stmt_metadata_mem_usage += stmt_global_info->ref_count_client;// *
|
||||
//((stmt_global_info->num_params * sizeof(MYSQL_BIND)) +
|
||||
//(stmt_global_info->num_columns * sizeof(MYSQL_FIELD))) + 16; // ~16 bytes of memory utilized by global_stmt_id and stmt_id mappings
|
||||
|
||||
// backend
|
||||
prep_stmt_backend_mem_usage += stmt_global_info->ref_count_server;// *(sizeof(MYSQL_STMT) +
|
||||
//56// + //sizeof(MADB_STMT_EXTENSION)
|
||||
//(stmt_global_info->num_params * sizeof(MYSQL_BIND)) +
|
||||
//(stmt_global_info->num_columns * sizeof(MYSQL_FIELD)));
|
||||
}
|
||||
unlock();
|
||||
}
|
||||
|
||||
void PgSQL_STMT_Manager_v14::get_metrics(uint64_t *c_unique, uint64_t *c_total,
|
||||
uint64_t *stmt_max_stmt_id, uint64_t *cached,
|
||||
uint64_t *s_unique, uint64_t *s_total) {
|
||||
#ifdef DEBUG
|
||||
uint64_t c_u = 0;
|
||||
uint64_t c_t = 0;
|
||||
uint64_t m = 0;
|
||||
uint64_t c = 0;
|
||||
uint64_t s_u = 0;
|
||||
uint64_t s_t = 0;
|
||||
#endif
|
||||
pthread_rwlock_wrlock(&rwlock_);
|
||||
statuses.cached = map_stmt_id_to_info.size();
|
||||
statuses.c_unique = statuses.cached - num_stmt_with_ref_client_count_zero;
|
||||
statuses.s_unique = statuses.cached - num_stmt_with_ref_server_count_zero;
|
||||
#ifdef DEBUG
|
||||
for (std::map<uint64_t, PgSQL_STMT_Global_info *>::iterator it = map_stmt_id_to_info.begin();
|
||||
it != map_stmt_id_to_info.end(); ++it) {
|
||||
PgSQL_STMT_Global_info *a = it->second;
|
||||
c++;
|
||||
if (a->ref_count_client) {
|
||||
c_u++;
|
||||
c_t += a->ref_count_client;
|
||||
}
|
||||
if (a->ref_count_server) {
|
||||
s_u++;
|
||||
s_t += a->ref_count_server;
|
||||
}
|
||||
if (it->first > m) {
|
||||
m = it->first;
|
||||
}
|
||||
}
|
||||
assert (c_u == statuses.c_unique);
|
||||
assert (c_t == statuses.c_total);
|
||||
assert (c == statuses.cached);
|
||||
assert (s_t == statuses.s_total);
|
||||
assert (s_u == statuses.s_unique);
|
||||
*stmt_max_stmt_id = m;
|
||||
#endif
|
||||
*stmt_max_stmt_id = next_statement_id; // this is max stmt_id, no matter if in used or not
|
||||
*c_unique = statuses.c_unique;
|
||||
*c_total = statuses.c_total;
|
||||
*cached = statuses.cached;
|
||||
*s_total = statuses.s_total;
|
||||
*s_unique = statuses.s_unique;
|
||||
pthread_rwlock_unlock(&rwlock_);
|
||||
}
|
||||
|
||||
|
||||
class PS_global_stats {
|
||||
public:
|
||||
uint64_t statement_id;
|
||||
char *username;
|
||||
char *schemaname;
|
||||
uint64_t digest;
|
||||
unsigned long long ref_count_client;
|
||||
unsigned long long ref_count_server;
|
||||
char *query;
|
||||
uint64_t num_columns;
|
||||
uint64_t num_params;
|
||||
PS_global_stats(uint64_t stmt_id, char *s, char *u, uint64_t d, char *q, unsigned long long ref_c, unsigned long long ref_s, uint64_t columns, uint64_t params) {
|
||||
statement_id = stmt_id;
|
||||
digest=d;
|
||||
query=strndup(q, mysql_thread___query_digests_max_digest_length);
|
||||
username=strdup(u);
|
||||
schemaname=strdup(s);
|
||||
ref_count_client = ref_c;
|
||||
ref_count_server = ref_s;
|
||||
num_columns = columns;
|
||||
num_params = params;
|
||||
}
|
||||
~PS_global_stats() {
|
||||
if (query) {
|
||||
free(query);
|
||||
query=NULL;
|
||||
}
|
||||
if (username) {
|
||||
free(username);
|
||||
username=NULL;
|
||||
}
|
||||
if (schemaname) {
|
||||
free(schemaname);
|
||||
schemaname=NULL;
|
||||
}
|
||||
}
|
||||
char **get_row() {
|
||||
char buf[128];
|
||||
char **pta=(char **)malloc(sizeof(char *)*PS_GLOBAL_STATUS_FIELD_NUM);
|
||||
sprintf(buf,"%lu",statement_id);
|
||||
pta[0]=strdup(buf);
|
||||
assert(schemaname);
|
||||
pta[1]=strdup(schemaname);
|
||||
assert(username);
|
||||
pta[2]=strdup(username);
|
||||
|
||||
sprintf(buf,"0x%016llX", (long long unsigned int)digest);
|
||||
pta[3]=strdup(buf);
|
||||
|
||||
assert(query);
|
||||
pta[4]=strdup(query);
|
||||
sprintf(buf,"%llu",ref_count_client);
|
||||
pta[5]=strdup(buf);
|
||||
sprintf(buf,"%llu",ref_count_server);
|
||||
pta[6]=strdup(buf);
|
||||
sprintf(buf,"%lu",num_columns);
|
||||
pta[7]=strdup(buf);
|
||||
sprintf(buf,"%lu",num_params);
|
||||
pta[8]=strdup(buf);
|
||||
|
||||
return pta;
|
||||
}
|
||||
void free_row(char **pta) {
|
||||
int i;
|
||||
for (i=0;i<PS_GLOBAL_STATUS_FIELD_NUM;i++) {
|
||||
assert(pta[i]);
|
||||
free(pta[i]);
|
||||
}
|
||||
free(pta);
|
||||
}
|
||||
};
|
||||
|
||||
|
||||
SQLite3_result * PgSQL_STMT_Manager_v14::get_prepared_statements_global_infos() {
|
||||
proxy_debug(PROXY_DEBUG_MYSQL_QUERY_PROCESSOR, 4, "Dumping current prepared statements global info\n");
|
||||
SQLite3_result *result=new SQLite3_result(PS_GLOBAL_STATUS_FIELD_NUM);
|
||||
rdlock();
|
||||
result->add_column_definition(SQLITE_TEXT,"stmt_id");
|
||||
result->add_column_definition(SQLITE_TEXT,"schemaname");
|
||||
result->add_column_definition(SQLITE_TEXT,"username");
|
||||
result->add_column_definition(SQLITE_TEXT,"digest");
|
||||
result->add_column_definition(SQLITE_TEXT,"query");
|
||||
result->add_column_definition(SQLITE_TEXT,"ref_count_client");
|
||||
result->add_column_definition(SQLITE_TEXT,"ref_count_server");
|
||||
result->add_column_definition(SQLITE_TEXT,"num_columns");
|
||||
result->add_column_definition(SQLITE_TEXT,"num_params");
|
||||
for (std::map<uint64_t, PgSQL_STMT_Global_info *>::iterator it = map_stmt_id_to_info.begin();
|
||||
it != map_stmt_id_to_info.end(); ++it) {
|
||||
PgSQL_STMT_Global_info *a = it->second;
|
||||
PS_global_stats * pgs = new PS_global_stats(a->statement_id,
|
||||
a->schemaname, a->username,
|
||||
a->hash, a->query,
|
||||
a->ref_count_client, a->ref_count_server, 0, 0);
|
||||
char **pta = pgs->get_row();
|
||||
result->add_row(pta);
|
||||
pgs->free_row(pta);
|
||||
delete pgs;
|
||||
}
|
||||
unlock();
|
||||
return result;
|
||||
}
|
||||
Loading…
Reference in new issue