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:fix-time-limit
marcelb:for_fork
marcelb:generic
marcelb:queue
..
compare: marcelb:dev
Branches Tags
marcelb:dev
marcelb:fix-time-limit
marcelb:for_fork
marcelb:generic
marcelb:queue

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 all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats
  1. 4
      .vscode/settings.json
  2. 55
      README.md
  3. 111
      lib/mysql.hpp
  4. 155
      src/mysql.cpp
  5. 3
      test/compile.sh
  6. 90
      test/test.cpp

4
.vscode/settings.json vendored
Unescape View File

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

55
README.md
Unescape 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
Unescape 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

155
src/mysql.cpp
Unescape 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;
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
lock_guard<mutex> lock(io);
for (uint32_t i=0; i<pool_size; i++) {
Connection* connection = create_connection();
connection_pool.push(connection);
}
}
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
run_tloop = false;
tloop_future.get();
#endif
if (tloop_type == time_loop_type::internal) {
run_tloop = false;
tloop_future.get();
} 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
Unescape 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
Unescape View File

@ -4,78 +4,34 @@
using namespace std;
using namespace chrono;
#include "../../asynco/lib/asynco.hpp"
#include "../../asynco/lib/timers.hpp"
using namespace marcelb::asynco;
#include "../lib/mysql.hpp"
using namespace marcelb::mysql;
#include "../../asynco/lib/asynco.hpp"
#include "../../asynco/lib/timers.hpp"
using namespace marcelb::asynco;
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;

Write Preview
Loading…
Cancel
Save