You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
proxysql/lib/Query_Cache.cpp

585 lines
16 KiB

#include <prometheus/counter.h>
#include "btree_map.h"
#include "proxysql.h"
#include "cpp.h"
#include "query_cache.hpp"
#include "proxysql_atomic.h"
#include "SpookyV2.h"
#include "prometheus_helpers.h"
#define THR_UPDATE_CNT(__a, __b, __c, __d) \
do {\
__a+=__c; \
if (__a>=__d) { \
__sync_fetch_and_add(&__b, __a - __a % __d); __a = __a % __d; \
} \
} while(0)
#define THR_DECREASE_CNT(__a, __b, __c, __d) \
do {\
__a+=__c; \
if (__a>=__d) { \
__sync_fetch_and_sub(&__b, __a - __a % __d); __a = __a % __d; \
} \
} while(0)
#ifdef DEBUG
#define DEB "_DEBUG"
#else
#define DEB ""
#endif /* DEBUG */
#define QUERY_CACHE_VERSION "1.2.0905" DEB
#define PROXYSQL_QC_PTHREAD_MUTEX
extern MySQL_Threads_Handler *GloMTH;
typedef btree::btree_map<uint64_t, QC_entry_t *> BtMap_cache;
class KV_BtreeArray {
private:
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_t lock;
#else
rwlock_t lock;
#endif
BtMap_cache bt_map;
PtrArray *ptrArray;
uint64_t purgeChunkSize;
uint64_t purgeIdx;
bool __insert(uint64_t, void *);
uint64_t freeable_memory;
public:
uint64_t tottopurge;
KV_BtreeArray();
~KV_BtreeArray();
uint64_t get_data_size();
void purge_some(unsigned long long, bool);
int cnt();
bool replace(uint64_t key, QC_entry_t *entry);
QC_entry_t *lookup(uint64_t key);
void empty();
};
__thread uint64_t __thr_cntSet=0;
__thread uint64_t __thr_cntGet=0;
__thread uint64_t __thr_cntGetOK=0;
__thread uint64_t __thr_dataIN=0;
__thread uint64_t __thr_dataOUT=0;
__thread uint64_t __thr_num_entries=0;
__thread uint64_t __thr_num_deleted=0;
__thread uint64_t __thr_size_values=0;
//__thread uint64_t __thr_freeable_memory=0;
#define DEFAULT_SQC_size 4*1024*1024
static uint64_t Glo_cntSet=0;
static uint64_t Glo_cntGet=0;
static uint64_t Glo_cntGetOK=0;
static uint64_t Glo_num_entries=0;
static uint64_t Glo_dataIN=0;
static uint64_t Glo_dataOUT=0;
static uint64_t Glo_cntPurge=0;
static uint64_t Glo_size_values=0;
static uint64_t Glo_total_freed_memory;
KV_BtreeArray::KV_BtreeArray() {
freeable_memory=0;
tottopurge=0;
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_init(&lock, NULL);
#else
spinlock_rwlock_init(&lock);
#endif
ptrArray = new PtrArray;
};
KV_BtreeArray::~KV_BtreeArray() {
proxy_debug(PROXY_DEBUG_QUERY_CACHE, 3, "Size of KVBtreeArray:%d , ptrArray:%llu\n", cnt() , ptrArray->len);
empty();
QC_entry_t *qce=NULL;
while (ptrArray->len) {
qce=(QC_entry_t *)ptrArray->remove_index_fast(0);
free(qce->value);
free(qce);
}
delete ptrArray;
};
uint64_t KV_BtreeArray::get_data_size() {
uint64_t r = __sync_fetch_and_add(&Glo_num_entries,0) * (sizeof(QC_entry_t)+sizeof(QC_entry_t *)*2+sizeof(uint64_t)*2); // + __sync_fetch_and_add(&Glo_size_values,0) ;
return r;
};
void KV_BtreeArray::purge_some(unsigned long long QCnow_ms, bool aggressive) {
uint64_t ret=0, i, _size=0;
QC_entry_t *qce;
unsigned long long access_ms_min=0;
unsigned long long access_ms_max=0;
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_rdlock(&lock);
#else
spin_rdlock(&lock);
#endif
for (i=0; i<ptrArray->len;i++) {
qce=(QC_entry_t *)ptrArray->index(i);
if (aggressive) { // we have been asked to do aggressive purging
if (access_ms_min==0) {
access_ms_min = qce->access_ms;
} else {
if (access_ms_min > qce->access_ms) {
access_ms_min = qce->access_ms;
}
}
if (access_ms_max==0) {
access_ms_max = qce->access_ms;
} else {
if (access_ms_max < qce->access_ms) {
access_ms_max = qce->access_ms;
}
}
} else { // no aggresssive purging , legacy algorithm
if (qce->expire_ms==EXPIRE_DROPIT || qce->expire_ms<QCnow_ms) {
ret++;
_size+=qce->length;
}
}
}
freeable_memory=_size;
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_unlock(&lock);
#else
spin_rdunlock(&lock);
#endif
bool cond_freeable_memory=false;
if (aggressive==false) {
uint64_t total_freeable_memory=0;
total_freeable_memory=freeable_memory + ret * (sizeof(QC_entry_t)+sizeof(QC_entry_t *)*2+sizeof(uint64_t)*2);
if ( total_freeable_memory > get_data_size()*0.01 ) {
cond_freeable_memory=true; // there is memory that can be freed
}
}
//if ( freeable_memory + ret * (sizeof(QC_entry_t) > get_data_size()*0.01) {
if ( aggressive || cond_freeable_memory ) {
uint64_t removed_entries=0;
uint64_t freed_memory=0;
unsigned long long access_ms_lower_mark=0;
if (aggressive) {
access_ms_lower_mark=access_ms_min+(access_ms_max-access_ms_min)*0.1; // hardcoded for now. Remove the entries with access time in the 10% range closest to access_ms_min
}
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_wrlock(&lock);
#else
spin_wrlock(&lock);
#endif
for (i=0; i<ptrArray->len;i++) {
qce=(QC_entry_t *)ptrArray->index(i);
bool drop_entry=false;
if (__sync_fetch_and_add(&qce->ref_count,0)<=1) { // currently not in use
if (qce->expire_ms==EXPIRE_DROPIT || qce->expire_ms<QCnow_ms) { //legacy algorithm
drop_entry=true;
}
if (aggressive) { // we have been asked to do aggressive purging
if (drop_entry==false) { // if the entry is already marked to be dropped, no further check
if (qce->access_ms < access_ms_lower_mark) {
drop_entry=true;
}
}
}
}
if (drop_entry) {
qce=(QC_entry_t *)ptrArray->remove_index_fast(i);
btree::btree_map<uint64_t, QC_entry_t *>::iterator lookup;
lookup = bt_map.find(qce->key);
if (lookup != bt_map.end()) {
bt_map.erase(lookup);
}
i--;
freed_memory+=qce->length;
removed_entries++;
free(qce->value);
free(qce);
}
}
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_unlock(&lock);
#else
spin_wrunlock(&lock);
#endif
THR_DECREASE_CNT(__thr_num_deleted,Glo_num_entries,removed_entries,1);
if (removed_entries) {
__sync_fetch_and_add(&Glo_total_freed_memory,freed_memory);
__sync_fetch_and_sub(&Glo_size_values,freed_memory);
__sync_fetch_and_add(&Glo_cntPurge,removed_entries);
}
}
};
int KV_BtreeArray::cnt() {
return bt_map.size();
};
bool KV_BtreeArray::replace(uint64_t key, QC_entry_t *entry) {
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_wrlock(&lock);
#else
spin_wrlock(&lock);
#endif
THR_UPDATE_CNT(__thr_cntSet,Glo_cntSet,1,1);
THR_UPDATE_CNT(__thr_size_values,Glo_size_values,entry->length,1);
THR_UPDATE_CNT(__thr_dataIN,Glo_dataIN,entry->length,1);
THR_UPDATE_CNT(__thr_num_entries,Glo_num_entries,1,1);
entry->ref_count=1;
ptrArray->add(entry);
btree::btree_map<uint64_t, QC_entry_t *>::iterator lookup;
lookup = bt_map.find(key);
if (lookup != bt_map.end()) {
lookup->second->expire_ms=EXPIRE_DROPIT;
__sync_fetch_and_sub(&lookup->second->ref_count,1);
bt_map.erase(lookup);
}
bt_map.insert(std::make_pair(key,entry));
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_unlock(&lock);
#else
spin_wrunlock(&lock);
#endif
return true;
}
QC_entry_t * KV_BtreeArray::lookup(uint64_t key) {
QC_entry_t *entry=NULL;
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_rdlock(&lock);
#else
spin_rdlock(&lock);
#endif
THR_UPDATE_CNT(__thr_cntGet,Glo_cntGet,1,1);
btree::btree_map<uint64_t, QC_entry_t *>::iterator lookup;
lookup = bt_map.find(key);
if (lookup != bt_map.end()) {
entry=lookup->second;
__sync_fetch_and_add(&entry->ref_count,1);
}
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_unlock(&lock);
#else
spin_rdunlock(&lock);
#endif
return entry;
};
void KV_BtreeArray::empty() {
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_wrlock(&lock);
#else
spin_wrlock(&lock);
#endif
btree::btree_map<uint64_t, QC_entry_t *>::iterator lookup;
while (bt_map.size()) {
lookup = bt_map.begin();
if ( lookup != bt_map.end() ) {
lookup->second->expire_ms=EXPIRE_DROPIT;
bt_map.erase(lookup);
}
}
#ifdef PROXYSQL_QC_PTHREAD_MUTEX
pthread_rwlock_unlock(&lock);
#else
spin_wrunlock(&lock);
#endif
};
using metric_name = std::string;
using metric_help = std::string;
using metric_tags = std::map<std::string, std::string>;
using qc_counter_tuple =
std::tuple<
p_qc_counter::metric,
metric_name,
metric_help,
metric_tags
>;
using qc_gauge_tuple =
std::tuple<
p_qc_gauge::metric,
metric_name,
metric_help,
metric_tags
>;
using qc_counter_vector = std::vector<qc_counter_tuple>;
using qc_gauge_vector = std::vector<qc_gauge_tuple>;
const std::tuple<qc_counter_vector, qc_gauge_vector>
qc_metrics_map = std::make_tuple(
qc_counter_vector {
std::make_tuple (
p_qc_counter::query_cache_count_get,
"proxysql_query_cache_count_get",
"Number of read requests.",
metric_tags {}
),
std::make_tuple (
p_qc_counter::query_cache_count_get_ok,
"proxysql_query_cache_count_get_ok",
"Number of successful read requests.",
metric_tags {}
),
std::make_tuple (
p_qc_counter::query_cache_count_set,
"proxysql_query_cache_count_set",
"Number of write requests.",
metric_tags {}
),
std::make_tuple (
p_qc_counter::query_cache_bytes_in,
"proxysql_query_cache_bytes_in",
"Number of bytes sent into the Query Cache.",
metric_tags {}
),
std::make_tuple (
p_qc_counter::query_cache_bytes_out,
"proxysql_query_cache_bytes_out",
"Number of bytes read from the Query Cache.",
metric_tags {}
),
std::make_tuple (
p_qc_counter::query_cache_purged,
"proxysql_query_cache_purged",
"Number of entries purged by the Query Cache due to TTL expiration.",
metric_tags {}
),
std::make_tuple (
p_qc_counter::query_cache_entries,
"proxysql_query_cache_entries",
"Number of entries currently stored in the query cache.",
metric_tags {}
)
},
qc_gauge_vector {
std::make_tuple (
p_qc_gauge::query_cache_memory_bytes,
"proxysql_query_cache_memory_bytes",
"Memory currently used by the query cache (more details later).",
metric_tags {}
)
}
);
uint64_t Query_Cache::get_data_size_total() {
int r=0;
int i;
for (i=0; i<SHARED_QUERY_CACHE_HASH_TABLES; i++) {
r+=KVs[i]->get_data_size();
}
r += __sync_fetch_and_add(&Glo_size_values,0);
return r;
};
unsigned int Query_Cache::current_used_memory_pct() {
uint64_t cur_size=get_data_size_total();
float pctf = (float) cur_size*100/max_memory_size;
if (pctf > 100) return 100;
int pct=pctf;
return pct;
}
Query_Cache::Query_Cache() {
#ifdef DEBUG
if (glovars.has_debug==false) {
#else
if (glovars.has_debug==true) {
#endif /* DEBUG */
perror("Incompatible debugging version");
exit(EXIT_FAILURE);
}
for (int i=0; i<SHARED_QUERY_CACHE_HASH_TABLES; i++) {
KVs[i]=new KV_BtreeArray();
}
QCnow_ms=monotonic_time()/1000;
size=SHARED_QUERY_CACHE_HASH_TABLES;
shutdown=0;
purge_loop_time=DEFAULT_purge_loop_time;
purge_total_time=DEFAULT_purge_total_time;
purge_threshold_pct_min=DEFAULT_purge_threshold_pct_min;
purge_threshold_pct_max=DEFAULT_purge_threshold_pct_max;
max_memory_size=DEFAULT_SQC_size;
// Initialize prometheus metrics
init_prometheus_counter_array<qc_metrics_map_idx, p_qc_counter>(qc_metrics_map, this->metrics.p_counter_array);
init_prometheus_gauge_array<qc_metrics_map_idx, p_qc_gauge>(qc_metrics_map, this->metrics.p_gauge_array);
};
void Query_Cache::p_update_metrics() {
this->metrics.p_gauge_array[p_qc_gauge::query_cache_memory_bytes]->Set(get_data_size_total());
p_update_counter(this->metrics.p_counter_array[p_qc_counter::query_cache_count_get], Glo_cntGet);
p_update_counter(this->metrics.p_counter_array[p_qc_counter::query_cache_count_get_ok], Glo_cntGetOK);
p_update_counter(this->metrics.p_counter_array[p_qc_counter::query_cache_count_set], Glo_cntSet);
p_update_counter(this->metrics.p_counter_array[p_qc_counter::query_cache_bytes_in], Glo_dataIN);
p_update_counter(this->metrics.p_counter_array[p_qc_counter::query_cache_bytes_out], Glo_dataOUT);
p_update_counter(this->metrics.p_counter_array[p_qc_counter::query_cache_purged], Glo_cntPurge);
p_update_counter(this->metrics.p_counter_array[p_qc_counter::query_cache_entries], Glo_num_entries);
}
void Query_Cache::print_version() {
fprintf(stderr,"In memory Standard Query Cache (SQC) rev. %s -- %s -- %s\n", QUERY_CACHE_VERSION, __FILE__, __TIMESTAMP__);
};
Query_Cache::~Query_Cache() {
unsigned int i;
for (i=0; i<SHARED_QUERY_CACHE_HASH_TABLES; i++) {
delete KVs[i];
}
};
unsigned char * Query_Cache::get(uint64_t user_hash, const unsigned char *kp, const uint32_t kl, uint32_t *lv, unsigned long long curtime_ms, unsigned long long cache_ttl) {
unsigned char *result=NULL;
uint64_t hk=SpookyHash::Hash64(kp, kl, user_hash);
unsigned char i=hk%SHARED_QUERY_CACHE_HASH_TABLES;
QC_entry_t *entry=KVs[i]->lookup(hk);
if (entry!=NULL) {
unsigned long long t=curtime_ms;
if (entry->expire_ms > t && entry->create_ms + cache_ttl > t) {
THR_UPDATE_CNT(__thr_cntGetOK,Glo_cntGetOK,1,1);
THR_UPDATE_CNT(__thr_dataOUT,Glo_dataOUT,entry->length,1);
result=(unsigned char *)malloc(entry->length);
memcpy(result,entry->value,entry->length);
*lv=entry->length;
if (t > entry->access_ms) entry->access_ms=t;
}
__sync_fetch_and_sub(&entry->ref_count,1);
}
return result;
}
bool Query_Cache::set(uint64_t user_hash, const unsigned char *kp, uint32_t kl, unsigned char *vp, uint32_t vl, unsigned long long create_ms, unsigned long long curtime_ms, unsigned long long expire_ms) {
QC_entry_t *entry = (QC_entry_t *)malloc(sizeof(QC_entry_t));
entry->klen=kl;
entry->length=vl;
entry->ref_count=0;
entry->value=(char *)malloc(vl);
memcpy(entry->value,vp,vl);
entry->self=entry;
entry->create_ms=create_ms;
entry->access_ms=curtime_ms;
entry->expire_ms=expire_ms;
uint64_t hk=SpookyHash::Hash64(kp, kl, user_hash);
unsigned char i=hk%SHARED_QUERY_CACHE_HASH_TABLES;
entry->key=hk;
KVs[i]->replace(hk, entry);
return true;
}
uint64_t Query_Cache::flush() {
int i;
uint64_t total_count=0;
for (i=0; i<SHARED_QUERY_CACHE_HASH_TABLES; i++) {
total_count+=KVs[i]->cnt();
KVs[i]->empty();
}
return total_count;
};
void * Query_Cache::purgeHash_thread(void *) {
unsigned int i;
unsigned int MySQL_Monitor__thread_MySQL_Thread_Variables_version;
MySQL_Thread * mysql_thr = new MySQL_Thread();
MySQL_Monitor__thread_MySQL_Thread_Variables_version=GloMTH->get_global_version();
mysql_thr->refresh_variables();
max_memory_size = (uint64_t) mysql_thread___query_cache_size_MB*1024*1024;
while (shutdown==0) {
usleep(purge_loop_time);
unsigned long long t=monotonic_time()/1000;
QCnow_ms=t;
unsigned int glover=GloMTH->get_global_version();
if (GloMTH) {
if (MySQL_Monitor__thread_MySQL_Thread_Variables_version < glover ) {
MySQL_Monitor__thread_MySQL_Thread_Variables_version=glover;
mysql_thr->refresh_variables();
max_memory_size = (uint64_t) mysql_thread___query_cache_size_MB*1024*1024;
}
}
unsigned int curr_pct=current_used_memory_pct();
if (curr_pct < purge_threshold_pct_min ) continue;
for (i=0; i<SHARED_QUERY_CACHE_HASH_TABLES; i++) {
KVs[i]->purge_some(QCnow_ms, (curr_pct > purge_threshold_pct_max));
}
}
delete mysql_thr;
return NULL;
};
SQLite3_result * Query_Cache::SQL3_getStats() {
const int colnum=2;
char buf[256];
char **pta=(char **)malloc(sizeof(char *)*colnum);
//Get_Memory_Stats();
SQLite3_result *result=new SQLite3_result(colnum);
result->add_column_definition(SQLITE_TEXT,"Variable_Name");
result->add_column_definition(SQLITE_TEXT,"Variable_Value");
// NOTE: as there is no string copy, we do NOT free pta[0] and pta[1]
{ // Used Memoery
pta[0]=(char *)"Query_Cache_Memory_bytes";
sprintf(buf,"%lu", get_data_size_total());
pta[1]=buf;
result->add_row(pta);
}
{ // Glo_cntGet
pta[0]=(char *)"Query_Cache_count_GET";
sprintf(buf,"%lu", Glo_cntGet);
pta[1]=buf;
result->add_row(pta);
}
{ // Glo_cntGetOK
pta[0]=(char *)"Query_Cache_count_GET_OK";
sprintf(buf,"%lu", Glo_cntGetOK);
pta[1]=buf;
result->add_row(pta);
}
{ // Glo_cntSet
pta[0]=(char *)"Query_Cache_count_SET";
sprintf(buf,"%lu", Glo_cntSet);
pta[1]=buf;
result->add_row(pta);
}
{ // Glo_dataIN
pta[0]=(char *)"Query_Cache_bytes_IN";
sprintf(buf,"%lu", Glo_dataIN);
pta[1]=buf;
result->add_row(pta);
}
{ // Glo_dataOUT
pta[0]=(char *)"Query_Cache_bytes_OUT";
sprintf(buf,"%lu", Glo_dataOUT);
pta[1]=buf;
result->add_row(pta);
}
{ // Glo_cntPurge
pta[0]=(char *)"Query_Cache_Purged";
sprintf(buf,"%lu", Glo_cntPurge);
pta[1]=buf;
result->add_row(pta);
}
{ // Glo_num_entries
pta[0]=(char *)"Query_Cache_Entries";
sprintf(buf,"%lu", Glo_num_entries);
pta[1]=buf;
result->add_row(pta);
}
free(pta);
return result;
}