marcelb/mySQL
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

Compare commits

base: marcelb:for_fork
Branches Tags
marcelb:dev
marcelb:for_fork
marcelb:fix-time-limit
marcelb:generic
marcelb:queue
marcelb:v0.9
marcelb:v0.8
marcelb:v0.7
marcelb:v0.6_fix1
marcelb:v0.6
marcelb:v0.5_queue
marcelb:v0.4_pool
marcelb:v0.3_pool
marcelb:v0.3_beta
marcelb:v0.2_beta
marcelb:v0.1_beta
..
compare: marcelb:dev
Branches Tags
marcelb:for_fork
marcelb:fix-time-limit
marcelb:dev
marcelb:generic
marcelb:queue
marcelb:v0.9
marcelb:v0.8
marcelb:v0.7
marcelb:v0.6_fix1
marcelb:v0.6
marcelb:v0.5_queue
marcelb:v0.4_pool
marcelb:v0.3_pool
marcelb:v0.3_beta
marcelb:v0.2_beta
marcelb:v0.1_beta

No commits in common. "for_fork" and "dev" have entirely different histories.
for_fork ... dev

6 changed files with 144 additions and 274 deletions
Show Stats Expand all files Collapse all files

4
.vscode/settings.json vendored
View File

@ -61,8 +61,6 @@
"cinttypes": "cpp",
"typeinfo": "cpp",
"any": "cpp",
"variant": "cpp",
"*.ipp": "cpp",
"bitset": "cpp"
"variant": "cpp"
}
}

55
README.md
View File

@ -13,7 +13,7 @@ A small framework for basic MySQL database operations via MySQL/Connector++
- Response object
- Thread safe
- Exceptions and log error callback
- Support my Asynco wrapper around Boost ASIO and support threads
- Can use external time loop for connection management
## Installation
@ -22,11 +22,6 @@ First install dependency MySQL/Connector++
```
sudo apt install libmysqlcppconn-dev
```
If you are going to use with an Asynco wrapper, download the archive from the profile and install the Boost dependencies
```
sudo apt install libboost-all-dev
```
Just download the latest release and unzip it into your project. You can turn it on with:
@ -34,15 +29,7 @@ Just download the latest release and unzip it into your project. You can turn it
#include "mysql/lib/mysql.hpp"
using namespace marcelb;
```
Compiling
```
# use with my Asynco lib
g++ -DMYSQL_USE_ASYNCO test.cpp ../src/* ../../asynco/src/* -o test.o -lmysqlcppconn -lpthread
# or use without asnyco (in multithread)
g++ test.cpp ../src/* ../../asynco/src/* -o test.o -lmysqlcppconn -lpthread
```
## Usage
### Internal engine
@ -59,14 +46,6 @@ using namespace marcelb::mysql;
*/
MySQL mydb("tcp://192.168.2.10:3306", "user_nm", "passss", "my_db", 5);
mydb.on_error = [](const string& error) {
cout << error << endl;
};
mydb.on_connect = []() {
cout << "Init all pool connection done" << endl;
};
/**
* Use ------------------
*/ | |
@ -93,16 +72,34 @@ try { | | | |
}
```
### Run async with Asynco
### External engine
As I developed quite a few wrappers that have some internal thread, I realized that it was inefficient and made it possible to call the necessary functions periodically outside (one thread per whole application or timer (ASIO), or my asynco wrapper).
```c++
#include "../lib/mysql.hpp"
using namespace marcelb::mysql;
#include "../../asynco/lib/timers.hpp"
using namespace marcelb::asynco;
/**
* Init
*/
MySQL mydb("tcp://192.168.2.10:3306", "user_nm", "passss", "my_db", 5, time_loop_type::external);
periodic mysql_maintenance ( [&mydb] () {
mydb.periodic_maintenance();
}, MYSQL_PERIODIC_INTERNAL_TIME);
mydb.set_on_error( [](const string& error) {
cout << error << endl; // print or log
});
/**
* You can call multiple queries asynchronously
*/
auto a1 = async_ ( [&mydb] () {
auto a1 = atask ( [&mydb] () {
try {
auto response = mydb.exec<int,string>("SELECT id,domain FROM records WHERE enabled = 1;");
for (auto row : response) {
@ -113,7 +110,7 @@ auto a1 = async_ ( [&mydb] () {
}
});
auto a2 = async_ ( [&mydb] () {
auto a2 = atask ( [&mydb] () {
try {
auto response = mydb.exec<string,string>("SELECT zonename,auth_key FROM zones;");
for (auto row : response) {
@ -124,7 +121,7 @@ auto a2 = async_ ( [&mydb] () {
}
});
auto a3 = async_ ( [&mydb] () {
auto a3 = atask ( [&mydb] () {
try {
auto response = mydb.exec<string,string>("SELECT username,email FROM users WHERE enabled = 1;");
for (auto row : response) {
@ -135,9 +132,9 @@ auto a3 = async_ ( [&mydb] () {
}
});
await(a1);
await(a2);
await(a3);
wait(a1);
wait(a2);
wait(a3);
```
## License

111
lib/mysql.hpp
View File

@ -8,8 +8,7 @@
#include <string>
#include <vector>
#include <tuple>
#include <chrono>
using namespace std;
#include "ctime"
#include <mysql_driver.h>
#include <mysql_connection.h>
@ -18,18 +17,13 @@ using namespace std;
#include <cppconn/statement.h>
#include <cppconn/prepared_statement.h>
#include <cppconn/resultset.h>
using namespace sql;
using namespace mysql;
#ifdef MYSQL_USE_ASYNCO
#include "../../asynco/lib/asynco.hpp"
#include "../../asynco/lib/timers.hpp"
using namespace marcelb::asynco;
#endif
#define unlimited 0
#define reconnectSleep 1000 // in us
#define reconnectSleep 10000 // in us
using namespace std;
using namespace sql;
using namespace mysql;
namespace marcelb {
namespace mysql {
@ -37,9 +31,18 @@ namespace mysql {
/**
*
*/
#define MYSQL_PERIODIC_INTERNAL_TIME 5000
#define MYSQL_MAX_CONNECTION_INIT_SAME_TIME 10
#define MYSQL_DELAYED_CONNECT_TIME 1000
#define MYSQL_PERIODIC_INTERNAL_TIME 1000
/**
* An enumeration of how periodic functions will be run
* internal - run periodic_maintenance() i new thread
* external - expects periodic_maintenance() to be run periodically outside the library
*
*/
enum class time_loop_type {
internal,
external
};
/**
* A class for creating sql responses
@ -100,16 +103,10 @@ class MySQL {
queue<Connection*> connection_pool;
string path, username, password, database;
uint32_t pool_size;
const uint32_t max_t = thread::hardware_concurrency();
uint32_t t = 1;
#ifdef MYSQL_USE_ASYNCO
periodic p_loop;
// delayed d_connect;
#else
future<void> tloop_future, connect_f;
bool run_tloop = true;
#endif
future<void> tloop_future;
time_loop_type tloop_type;
time_t last_loop_time;
/**
* Open one database server connection
@ -156,59 +153,10 @@ class MySQL {
* Internal tloop periodic
*/
void _tloop(uint32_t b, uint32_t e);
/**
*
*/
vector<pair<int, int>> divide_and_conquer(int n, int t) {
vector<pair<int, int>> pairs;
int part_size = n / t;
int residue = n % t; // Ostatak koji treba rasporediti
for (int i = 0; i < t; i++) {
int start = i * part_size + std::min(i, residue);
int end = start + part_size + (i < residue ? 1 : 0);
pairs.push_back(make_pair(start, end));
}
return pairs;
}
/**
* Automatically adjusts
*/
void auto_adjusts(const uint32_t duration) {
if (duration > (MYSQL_PERIODIC_INTERNAL_TIME*2)/3) {
if (t < max_t) {
t++; // povećaj broj asinkronih zadataka drugi put
}
else if (on_error) {
on_error("Periodic maintenance takes too long");
}
}
else if (duration < MYSQL_PERIODIC_INTERNAL_TIME/3 && t > 1) {
t--; // smanji nema potrebe
}
}
#ifndef MYSQL_USE_ASYNCO
/**
*
*/
uint64_t timestamp() {
return chrono::duration_cast<chrono::milliseconds>(
chrono::system_clock::now().time_since_epoch()
).count();
};
#endif
void _tloop();
public:
function<void(const string&)> on_error;
function<void()> on_connect;
uint32_t connect_trys = 3;
@ -218,7 +166,7 @@ public:
* username, password, database name,
* and number of active connections (optional)
*/
MySQL(const string _path, const string _username, const string _password, const string _db, const uint32_t _available = 1);
MySQL(const string _path, const string _username, const string _password, const string _db, const uint32_t _available = 1, const time_loop_type _engine_type = time_loop_type::internal);
/**
* Execute the SQL statement
@ -257,14 +205,23 @@ public:
stmt->close();
delete stmt;
release_connection(connection);
} catch (sql::SQLException& e) {
throw runtime_error(e.what());
// cerr << "SQLState: " << e.getSQLState() << endl;
// cerr << "Error code: " << e.getErrorCode() << endl;
// std::cerr << "SQLState: " << e.getSQLState() << std::endl;
// std::cerr << "Error code: " << e.getErrorCode() << std::endl;
}
return result;
}
/**
* If you are using an external periodic motor,
* please call this function in it for proper operation at a certain time interval.
* You can use the default MYSQL_PERIODIC_INTERNAL_TIME
*/
void tloop();
/**
* Destruktor
* close all connections

143
src/mysql.cpp
View File

@ -1,69 +1,28 @@
#include "../lib/mysql.hpp"
marcelb::mysql::MySQL::MySQL(const string _path, const string _username, const string _password, const string _db, const uint32_t _available):
#ifdef MYSQL_USE_ASYNCO
p_loop(periodic( [&] () {
const uint64_t start = rtime_ms();
auto pairs = divide_and_conquer(connection_pool.size(), t);
vector<future<void>> fuve;
for (auto& pair : pairs) {
fuve.push_back(async_ ([&, pair](){
_tloop(pair.first, pair.second);
}));
}
for (auto& fu : fuve) {
try {
await_(fu);
} catch (...) {
if(on_error) {
on_error("Error in maintenance periodic loop.");
}
}
}
auto_adjusts(rtime_ms() - start);
}, MYSQL_PERIODIC_INTERNAL_TIME)),
#else
tloop_future (async(launch::async, [&](){
while (run_tloop) {
usleep(MYSQL_PERIODIC_INTERNAL_TIME*1000);
const uint64_t start = timestamp();
auto pairs = divide_and_conquer(connection_pool.size(), t);
vector<future<void>> fuve;
for (auto& pair : pairs) {
fuve.push_back(async (launch::async, [&, pair](){
_tloop(pair.first, pair.second);
}));
}
for (auto& fu : fuve) {
try {
fu.get();
} catch (...) {
if(on_error) {
on_error("Error in maintenance periodic loop.");
}
}
}
auto_adjusts(timestamp() - start);
}
return;
})),
#endif
path(_path),
username(_username),
password(_password),
database(_db),
pool_size(_available > 0 ? _available : 1) {
marcelb::mysql::MySQL::MySQL(const string _path, const string _username, const string _password, const string _db, const uint32_t _available, const time_loop_type _engine_type) {
path = _path;
username = _username;
password = _password;
database = _db;
pool_size = _available > 0 ? _available : 1;
tloop_type = _engine_type;
drv = get_mysql_driver_instance();
connect_pool();
if (tloop_type == time_loop_type::internal) {
tloop_future = async(launch::async, [&](){
while (run_tloop) {
usleep(MYSQL_PERIODIC_INTERNAL_TIME*1000);
_tloop();
}
return;
});
}
// set on initialization to avoid the error
last_loop_time = time(nullptr);
}
@ -71,6 +30,7 @@ Connection* marcelb::mysql::MySQL::create_connection() {
uint32_t trys = 0;
bool status = true;
Connection* new_con = NULL;
while (connect_trys == unlimited ? status : (trys <= connect_trys && status)) {
try {
Connection* con_can = drv->connect(path, username, password);
@ -96,28 +56,11 @@ Connection* marcelb::mysql::MySQL::create_connection() {
}
void marcelb::mysql::MySQL::connect_pool() {
auto connect_ = [&]() -> void {
const uint32_t tens = pool_size/MYSQL_MAX_CONNECTION_INIT_SAME_TIME;
lock_guard<mutex> lock(io);
for (uint32_t i=0; i<pool_size; i++) {
Connection* connection = create_connection();
{
lock_guard<mutex> lock(io);
connection_pool.push(connection);
}
if ((i+1)%tens == 0) {
usleep(MYSQL_DELAYED_CONNECT_TIME*1000);
}
}
if (on_connect) {
on_connect();
}
};
#ifdef MYSQL_USE_ASYNCO
async_ (connect_);
#else
connect_f = async (launch::async, connect_);
#endif
}
void marcelb::mysql::MySQL::disconnect_pool() {
@ -152,20 +95,23 @@ bool marcelb::mysql::MySQL::disconnect_connection(Connection* connection) {
return status;
}
void marcelb::mysql::MySQL::_tloop(uint32_t b, uint32_t e) {
for (size_t i=b; i<connection_pool.size() && i<e; i++) {
void marcelb::mysql::MySQL::_tloop() {
if (!run_tloop) {
return;
}
lock_guard<mutex> lock(io);
for (size_t i=0; i<connection_pool.size(); i++) {
try {
Connection *conn = nullptr; // provjeri ovdje svugdje možeš li koristiti release i ocupacy sada
conn = occupy_connection();
Connection *conn = connection_pool.front();
connection_pool.pop();
if (conn->isValid()) {
release_connection(conn);
connection_pool.push(conn);
} else {
if (!conn->isClosed()){
disconnect_connection(conn);
conn->close();
}
Connection *n_conn = create_connection();
release_connection(n_conn);
}
} catch (const SQLException &error) {
if (on_error) {
@ -173,9 +119,16 @@ void marcelb::mysql::MySQL::_tloop(uint32_t b, uint32_t e) {
}
}
}
last_loop_time = time(nullptr);
}
Connection* marcelb::mysql::MySQL::occupy_connection() {
if (last_loop_time + (MYSQL_PERIODIC_INTERNAL_TIME*3/1000) < time(nullptr)) {
if (on_error) {
on_error("The time loop is not executing properly");
}
}
unique_lock<mutex> lock(io);
while (connection_pool.empty()) {
condition.wait(lock);
@ -192,11 +145,23 @@ void marcelb::mysql::MySQL::release_connection(Connection* connection) {
}
marcelb::mysql::MySQL::~MySQL() {
#ifdef MYSQL_USE_ASYNCO
p_loop.stop();
#else
if (tloop_type == time_loop_type::internal) {
run_tloop = false;
tloop_future.get();
#endif
} else {
run_tloop = false;
}
disconnect_pool();
}
void marcelb::mysql::MySQL::tloop() {
if (tloop_type == time_loop_type::internal) {
if (on_error) {
on_error("Can't start external call tloop, internal is active!");
}
return;
}
_tloop();
}

3
test/compile.sh
View File

@ -1,2 +1 @@
g++ -DMYSQL_USE_ASYNCO test.cpp ../src/* ../../asynco/src/* -o test.o -lmysqlcppconn -lpthread
# g++ test.cpp ../src/* ../../asynco/src/* -o test.o -lmysqlcppconn -lpthread
g++ test.cpp ../src/* -o test.o -lmysqlcppconn -lpthread

90
test/test.cpp
View File

@ -4,78 +4,34 @@
using namespace std;
using namespace chrono;
#include "../lib/mysql.hpp"
using namespace marcelb::mysql;
#include "../../asynco/lib/asynco.hpp"
#include "../../asynco/lib/timers.hpp"
using namespace marcelb::asynco;
#include "../lib/mysql.hpp"
using namespace marcelb::mysql;
int main() {
auto inis = rtime_ms();
try {
const int n = 10;
// MySQL mydb("tcp://192.168.2.10:3306", "dinio", "H€r5elfInd1aH@nds", "dinio", 5, time_loop_type::internal);
MySQL mydb("tcp://bitelex.ddns.net:3306", "dinio", "H€r5elfInd1aH@nds", "dinio", n);
MySQL mydb("tcp://bitelex.ddns.net:3306", "dinio", "H€r5elfInd1aH@nds", "dinio", 5, time_loop_type::external);
// MySQL mydb("tcp://bitelex.ddns.net:3306", "dinio", "H€r5elfInd1aH@nds", "dinio", 5);
cout << "--------------init: " << rtime_ms() - inis << endl;
mydb.on_error = [](const string& error) {
cout << error << endl;
};
mydb.on_connect = []() {
cout << "Init all pool connection done" << endl;
};
// periodic mysql_tloop ( [&mydb] () {
// auto l_start = rtime_ms();
// vector<future<void>> to_wait;
// for (int i=0, old_i=0; i<n; old_i=i) {
// i += 5;
// to_wait.push_back( async_ ([&, i, old_i](){
// try {
// auto start = rtime_ms();
// mydb.tloop(old_i, i);
// cout << "old " << old_i << " i " << i << endl;
// cout << "loop--------------------------- nema error, trajalo: " << rtime_ms() - start << endl;
// } catch (...) {
// cout << "Bude neki error u loopu" << endl;
// }
// }));
// }
// async_ ([&](){
// try {
// auto start = rtime_ms();
// mydb.tloop(4, 8);
// cout << "loop--------------------------- nema error, trajalo: " << rtime_ms() - start << endl;
// } catch (...) {
// cout << "Bude neki error u loopu" << endl;
// }
// });
// async_ ([&](){
// try {
// auto start = rtime_ms();
// mydb.tloop(8, 12);
// cout << "loop--------------------------- nema error, trajalo: " << rtime_ms() - start << endl;
// } catch (...) {
// cout << "Bude neki error u loopu" << endl;
// }
// });
// for (auto& tw : to_wait) {
// wait (tw);
// }
// cout << "all loop !!!!!!!!!!!!!!1, trajalo: " << rtime_ms() - l_start << endl;
// }, 5000);
periodic mysql_tloop ( [&mydb] () {
cout << "loop---------------------------" << endl;
mydb.tloop();
}, MYSQL_PERIODIC_INTERNAL_TIME);
while (true) {
sleep(60);
sleep(5);
auto start = high_resolution_clock::now();
auto a1 = async_ ( [&mydb] () {
auto a1 = nonsync ( [&mydb] () {
try {
auto response = mydb.exec<int,string>("SELECT id,domain FROM records WHERE enabled = 1;");
cout << response.affected << " " << response.have_result << endl;
@ -94,7 +50,7 @@ while (true) {
}
});
auto a2 = async_ ( [&mydb] () {
auto a2 = nonsync ( [&mydb] () {
try {
auto response = mydb.exec<string,string>("SELECT zonename,auth_key FROM zones;");
cout << response.affected << " " << response.have_result << endl;
@ -113,7 +69,7 @@ while (true) {
}
});
auto a3 = async_ ( [&mydb] () {
auto a3 = nonsync ( [&mydb] () {
try {
auto response = mydb.exec<string,string>("SELECT username,email FROM users WHERE enabled = 1;");
cout << response.affected << " " << response.have_result << endl;
@ -132,7 +88,7 @@ while (true) {
}
});
auto a4 = async_ ( [&mydb] () {
auto a4 = nonsync ( [&mydb] () {
try {
auto response = mydb.exec<int,string>("SELECT id,domain FROM records WHERE enabled = 1;");
cout << response.affected << " " << response.have_result << endl;
@ -151,7 +107,7 @@ while (true) {
}
});
auto a5 = async_ ( [&mydb] () {
auto a5 = nonsync ( [&mydb] () {
try {
auto response = mydb.exec<string,string>("SELECT zonename,auth_key FROM zones;");
cout << response.affected << " " << response.have_result << endl;
@ -170,7 +126,7 @@ while (true) {
}
});
auto a6 = async_ ( [&mydb] () {
auto a6 = nonsync ( [&mydb] () {
try {
auto response = mydb.exec<string,string>("SELECT username,email FROM users WHERE enabled = 1;");
cout << response.affected << " " << response.have_result << endl;
@ -189,12 +145,12 @@ while (true) {
}
});
await_(a1);
await_(a2);
await_(a3);
await_(a4);
await_(a5);
await_(a6);
wait(a1);
wait(a2);
wait(a3);
wait(a4);
wait(a5);
wait(a6);
auto end = high_resolution_clock::now();
@ -207,9 +163,7 @@ while (true) {
} catch (const SQLException error) {
cout << error.what() << endl;
} catch (const exception& error) {
cout << error.what() << endl;
} catch (const string& error) {
} catch (const string error) {
cout << error << endl;
} catch (...) {
cout << "Jebi ga" << endl;