Compare commits

..

No commits in common. 'dev' and 'nonsync_wait' have entirely different histories.

  1. 1
      .gitignore
  2. 4
      .vscode/settings.json
  3. 22
      README.md
  4. 69
      lib/asynco.hpp
  5. 71
      lib/engine.hpp
  6. 124
      lib/timers.hpp
  7. 6
      lib/trigger.hpp
  8. 7
      src/engine.cpp
  9. 145
      src/timers.cpp
  10. 1
      test/compile.sh
  11. 67
      test/test.cpp

1
.gitignore vendored

@ -1,3 +1,2 @@
test/test test/test
test/*.txt test/*.txt
example

@ -69,8 +69,6 @@
"cinttypes": "cpp", "cinttypes": "cpp",
"typeindex": "cpp", "typeindex": "cpp",
"typeinfo": "cpp", "typeinfo": "cpp",
"variant": "cpp", "variant": "cpp"
"coroutine": "cpp",
"source_location": "cpp"
} }
} }

@ -77,28 +77,6 @@ int t = time1.expired();
// is it stopped // is it stopped
bool stoped = time1.stoped(); bool stoped = time1.stoped();
// If you don't want to save in a variable, but you want to start a timer, use these functions
// And you can also save them, they are only of the shared pointer type
auto d = Delayed( [](){
cout << "Delayed" << endl;
}, 2000);
auto p = Periodic( [](){
cout << "Periodic" << endl;
}, 700);
Periodic( [&] (){
cout << "Delayed expire " << d->expired() << endl;
cout << "Periodic ticks " << p->ticks() << endl;
cout << "Delayed stoped " << d->stoped() << endl;
cout << "Periodic stoped " << p->stoped() << endl;
}, 1000);
Delayed( [&](){
p->stop();
}, 10000);
``` ```
Make functions asynchronous Make functions asynchronous

@ -1,7 +1,7 @@
#ifndef _ASYNCO_ #ifndef _ASYNCO_
#define _ASYNCO_ #define _ASYNCO_
#include "engine.hpp" #include <boost/asio.hpp>
#include <iostream> #include <iostream>
using namespace std; using namespace std;
@ -9,6 +9,51 @@ using namespace std;
namespace marcelb { namespace marcelb {
namespace asynco { namespace asynco {
#define HW_CONCURRENCY_MINIMAL 4
/**
* Internal anonymous class for initializing the ASIO context and thread pool
* !!! It is anonymous to protect against use in the initialization of other objects of the same type !!!
*/
class {
public:
boost::asio::io_context io_context;
void run() {
for (auto& runner : runners) {
runner.join();
}
}
private:
unique_ptr<boost::asio::io_service::work> work { [&] () {
return new boost::asio::io_service::work(io_context);
} ()};
vector<thread> runners { [&] () {
vector<thread> _runs;
unsigned int num_of_runners;
#ifdef NUM_OF_RUNNERS
num_of_runners = NUM_OF_RUNNERS;
#else
num_of_runners = thread::hardware_concurrency();
if (num_of_runners < HW_CONCURRENCY_MINIMAL) {
num_of_runners = HW_CONCURRENCY_MINIMAL;
}
#endif
for (int i=0; i<num_of_runners; i++) {
_runs.push_back(thread ( [this] () {
io_context.run();
}));
}
return _runs;
} ()};
} _asynco_engine;
/** /**
* Run the function asynchronously * Run the function asynchronously
*/ */
@ -35,28 +80,6 @@ T await_(future<T>&& r) {
return move(r).get(); return move(r).get();
} }
/**
* Block until the asynchronous call completes or time expired
*/
template<typename T>
T await_(future<T>& r, uint64_t time) {
if (r.wait_for(chrono::milliseconds(time)) == std::future_status::timeout) {
throw runtime_error("Asynchronous execution timed out");
}
return r.get();
}
/**
* Block until the asynchronous call completes or time expired
*/
template<typename T>
T await_(future<T>&& r, uint64_t time) {
if (r.wait_for(chrono::milliseconds(time)) == std::future_status::timeout) {
throw runtime_error("Asynchronous execution timed out");
}
return move(r).get();
}
} }
} }

@ -1,71 +0,0 @@
#ifndef _ASYNCO_ENGINE_
#define _ASYNCO_ENGINE_
#include <vector>
#include <memory>
using namespace std;
#include <boost/asio.hpp>
namespace marcelb {
namespace asynco {
#define HW_CONCURRENCY_MINIMAL 4
/**
* Internal anonymous class for initializing the ASIO context and thread pool
* !!! It is anonymous to protect against use in the initialization of other objects of the same type !!!
*/
class Engine {
public:
boost::asio::io_context io_context;
void run() {
for (auto& runner : runners) {
runner.join();
}
}
private:
unique_ptr<boost::asio::io_service::work> work { [&] () {
return new boost::asio::io_service::work(io_context);
} ()};
vector<thread> runners { [&] () {
vector<thread> _runs;
unsigned int num_of_runners;
#ifdef NUM_OF_RUNNERS
num_of_runners = NUM_OF_RUNNERS;
#else
num_of_runners = thread::hardware_concurrency();
if (num_of_runners < HW_CONCURRENCY_MINIMAL) {
num_of_runners = HW_CONCURRENCY_MINIMAL;
}
#endif
for (int i=0; i<num_of_runners; i++) {
_runs.push_back(thread ( [this] () {
io_context.run();
}));
}
return _runs;
} ()};
};
extern Engine _asynco_engine;
}
}
#endif

@ -1,10 +1,12 @@
#ifndef _ASYNCO_TIMERS_ #ifndef _TIMERS_
#define _ASYNCO_TIMERS_ #define _TIMERS_
#include "asynco.hpp"
#include <chrono> #include <chrono>
using namespace std;
#include "asynco.hpp" using namespace std;
using namespace marcelb;
using namespace asynco;
namespace marcelb { namespace marcelb {
namespace asynco { namespace asynco {
@ -13,13 +15,21 @@ namespace asynco {
* Get the time in ms from the epoch * Get the time in ms from the epoch
*/ */
int64_t rtime_ms(); int64_t rtime_ms() {
return chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now()
.time_since_epoch())
.count();
}
/** /**
* Get the time in us from the epoch * Get the time in us from the epoch
*/ */
int64_t rtime_us(); int64_t rtime_us() {
return chrono::duration_cast<chrono::microseconds>(chrono::system_clock::now()
.time_since_epoch())
.count();
}
/** /**
* Core timer class for construct time async functions * Core timer class for construct time async functions
@ -35,40 +45,71 @@ class timer {
/** /**
* A method to assign a callback wrapper and a reinitialization algorithm * A method to assign a callback wrapper and a reinitialization algorithm
*/ */
void init(); void init() {
st.async_wait( [this] (const boost::system::error_code&) {
if (!_stop) {
callback();
if (repeate) {
st = boost::asio::steady_timer(_asynco_engine.io_context, boost::asio::chrono::milliseconds(time));
init();
}
_ticks++;
}
});
}
public: public:
/** /**
* The constructor creates the steady_timer and accompanying variables and runs a method to initialize the timer * The constructor creates the steady_timer and accompanying variables and runs a method to initialize the timer
*/ */
timer (function<void()> _callback, uint64_t _time, bool _repeate); timer (function<void()> _callback, uint64_t _time, bool _repeate) :
st(_asynco_engine.io_context, boost::asio::chrono::milliseconds(_time)),
_stop(false),
repeate(_repeate),
callback(_callback),
time(_time) {
init();
}
/** /**
* Stop timer * Stop timer
* The stop flag is set and timer remove it from the queue * The stop flag is set and timer remove it from the queue
*/ */
void stop(); void stop() {
_stop = true;
st.cancel();
}
/** /**
* Run callback now * Run callback now
* Forces the callback function to run independently of the timer * Forces the callback function to run independently of the timer
*/ */
void now(); void now() {
st.cancel();
}
/** /**
* Get the number of times the timer callback was runned * Get the number of times the timer callback was runned
*/ */
uint64_t ticks(); uint64_t ticks() {
return _ticks;
}
/** /**
* The logic status of the timer stop state * The logic status of the timer stop state
*/ */
bool stoped(); bool stoped() {
return _stop;
}
/** /**
* The destructor stops the timer * The destructor stops the timer
*/ */
~timer(); ~timer() {
stop();
}
}; };
/** /**
@ -78,37 +119,49 @@ class periodic {
shared_ptr<timer> _timer; shared_ptr<timer> _timer;
public: public:
/** /**
* Constructor initializes a shared pointer of type timer * Constructor initializes a shared pointer of type timer
*/ */
periodic(function<void()> callback, uint64_t time); periodic(function<void()> callback, uint64_t time) :
_timer(make_shared<timer> (callback, time, true)) {
}
/** /**
* Stop periodic * Stop periodic
* The stop flag is set and periodic remove it from the queue * The stop flag is set and periodic remove it from the queue
*/ */
void stop(); void stop() {
_timer->stop();
}
/** /**
* Run callback now * Run callback now
* Forces the callback function to run independently of the periodic * Forces the callback function to run independently of the periodic
*/ */
void now(); void now() {
_timer->now();
}
/** /**
* Get the number of times the periodic callback was runned * Get the number of times the periodic callback was runned
*/ */
uint64_t ticks(); uint64_t ticks() {
return _timer->ticks();
}
/** /**
* The logic status of the periodic stop state * The logic status of the periodic stop state
*/ */
bool stoped(); bool stoped() {
return _timer->stoped();
}
/** /**
* The destructor stops the periodic * The destructor stops the periodic
*/ */
~periodic(); ~periodic() {
stop();
}
}; };
/** /**
@ -118,43 +171,52 @@ class delayed {
shared_ptr<timer> _timer; shared_ptr<timer> _timer;
public: public:
/** /**
* Constructor initializes a shared pointer of type timer * Constructor initializes a shared pointer of type timer
*/ */
delayed(function<void()> callback, uint64_t time); delayed(function<void()> callback, uint64_t time) :
_timer(make_shared<timer> (callback, time, false)) {
}
/** /**
* Stop delayed * Stop delayed
* The stop flag is set and delayed remove it from the queue * The stop flag is set and delayed remove it from the queue
*/ */
void stop(); void stop() {
_timer->stop();
}
/** /**
* Run callback now * Run callback now
* Forces the callback function to run independently of the delayed * Forces the callback function to run independently of the delayed
*/ */
void now(); void now() {
_timer->now();
}
/** /**
* Get is the delayed callback runned * Get is the delayed callback runned
*/ */
bool expired(); bool expired() {
return bool(_timer->ticks());
}
/** /**
* The logic status of the delayed stop state * The logic status of the delayed stop state
*/ */
bool stoped(); bool stoped() {
return _timer->stoped();
}
/** /**
* The destructor stops the delayed * The destructor stops the delayed
*/ */
~delayed(); ~delayed() {
stop();
}
}; };
shared_ptr<periodic> Periodic(function<void()> callback, uint64_t time);
shared_ptr<delayed> Delayed(function<void()> callback, uint64_t time);
} }
} }

@ -1,5 +1,5 @@
#ifndef _ASYNCO_TRIGGER_ #ifndef _TRIGGER_
#define _ASYNCO_TRIGGER_ #define _TRIGGER_
#include <map> #include <map>
#include <vector> #include <vector>
@ -8,7 +8,7 @@
using namespace std; using namespace std;
#include "engine.hpp" #include "asynco.hpp"
namespace marcelb { namespace marcelb {
namespace asynco { namespace asynco {
namespace triggers { namespace triggers {

@ -1,7 +0,0 @@
#include "../lib/engine.hpp"
namespace marcelb::asynco {
Engine _asynco_engine;
};

@ -1,145 +0,0 @@
#include "../lib/timers.hpp"
namespace marcelb::asynco {
int64_t rtime_ms() {
return chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now()
.time_since_epoch())
.count();
}
int64_t rtime_us() {
return chrono::duration_cast<chrono::microseconds>(chrono::system_clock::now()
.time_since_epoch())
.count();
}
void timer::init() {
st.async_wait( [this] (const boost::system::error_code&) {
if (!_stop) {
callback();
if (repeate) {
st = boost::asio::steady_timer(_asynco_engine.io_context, boost::asio::chrono::milliseconds(time));
init();
}
_ticks++;
}
});
}
timer::timer (function<void()> _callback, uint64_t _time, bool _repeate) :
st(_asynco_engine.io_context, boost::asio::chrono::milliseconds(_time)),
_stop(false),
repeate(_repeate),
callback(_callback),
time(_time) {
init();
}
void timer::stop() {
_stop = true;
st.cancel();
}
void timer::now() {
st.cancel();
}
uint64_t timer::ticks() {
return _ticks;
}
bool timer::stoped() {
return _stop;
}
timer::~timer() {
stop();
}
periodic::periodic(function<void()> callback, uint64_t time) :
_timer(make_shared<timer> (callback, time, true)) {
}
void periodic::stop() {
_timer->stop();
}
void periodic::now() {
_timer->now();
}
uint64_t periodic::ticks() {
return _timer->ticks();
}
bool periodic::stoped() {
return _timer->stoped();
}
periodic::~periodic() {
stop();
}
delayed::delayed(function<void()> callback, uint64_t time) :
_timer(make_shared<timer> (callback, time, false)) {
}
void delayed::stop() {
_timer->stop();
}
void delayed::now() {
_timer->now();
}
bool delayed::expired() {
return bool(_timer->ticks());
}
bool delayed::stoped() {
return _timer->stoped();
}
delayed::~delayed() {
stop();
}
mutex p_io, d_io;
vector<shared_ptr<periodic>> periodic_calls_container;
vector<shared_ptr<delayed>> delayed_calls_container;
shared_ptr<periodic> Periodic(function<void()> callback, uint64_t time) {
shared_ptr<periodic> periodic_ptr(make_shared<periodic>(callback, time));
async_ ( [&, periodic_ptr](){
lock_guard<mutex> lock(p_io);
periodic_calls_container.push_back(periodic_ptr);
for (uint32_t i=0; i<periodic_calls_container.size(); i++) {
if (periodic_calls_container[i]->stoped()) {
periodic_calls_container.erase(periodic_calls_container.begin()+i);
i--;
}
}
});
return periodic_ptr;
}
shared_ptr<delayed> Delayed(function<void()> callback, uint64_t time) {
shared_ptr<delayed> delayed_ptr(make_shared<delayed>(callback, time));
async_ ( [&, delayed_ptr](){
lock_guard<mutex> lock(p_io);
delayed_calls_container.push_back(delayed_ptr);
for (uint32_t i=0; i<delayed_calls_container.size(); i++) {
if (delayed_calls_container[i]->stoped() || delayed_calls_container[i]->expired()) {
delayed_calls_container.erase(delayed_calls_container.begin()+i);
i--;
}
}
});
return delayed_ptr;
}
};

@ -1 +0,0 @@
g++ test.cpp ../src/* -o test

@ -69,9 +69,9 @@ int main () {
// --------------- TIME ASYNCHRONOUS FUNCTIONS -------------- // --------------- TIME ASYNCHRONOUS FUNCTIONS --------------
/** // /**
* Init periodic and delayed; clear periodic and delayed // * Init periodic and delayed; clear periodic and delayed
*/ // */
// periodic inter1 ([&]() { // periodic inter1 ([&]() {
// cout << "periodic prvi " << rtime_ms() - start << endl; // cout << "periodic prvi " << rtime_ms() - start << endl;
@ -129,30 +129,11 @@ int main () {
// cout << "nije isteko " << endl; // cout << "nije isteko " << endl;
// } // }
// auto d = Delayed( [](){ // // // ------------------------ MAKE FUNCTIONS ASYNCHRONOUS -------------------------
// cout << "Delayed" << endl;
// }, 2000);
// auto p = Periodic( [](){
// cout << "Periodic" << endl;
// }, 700);
// Periodic( [&] (){ // /**
// cout << "Delayed expire " << d->expired() << endl; // * Run an function asyncronic
// cout << "Periodic ticks " << p->ticks() << endl; // */
// cout << "Delayed stoped " << d->stoped() << endl;
// cout << "Periodic stoped " << p->stoped() << endl;
// }, 1000);
// Delayed( [&](){
// p->stop();
// }, 10000);
// // // // ------------------------ MAKE FUNCTIONS ASYNCHRONOUS -------------------------
// // /**
// // * Run an function asyncronic
// // */
// async_ ( []() { // async_ ( []() {
// sleep_for(2s); // only for simulate log duration function // sleep_for(2s); // only for simulate log duration function
@ -174,13 +155,13 @@ int main () {
// ); // );
// // async(launch::async, [] () { // async(launch::async, [] () {
// // cout << "Another thread in async style!" << endl; // cout << "Another thread in async style!" << endl;
// // }); // });
// // /** // /**
// // * Call class method // * Call class method
// // */ // */
// clm classes; // clm classes;
// async_ ( [&classes] () { // async_ ( [&classes] () {
@ -189,17 +170,17 @@ int main () {
// sleep(5); // sleep(5);
// // /** // /**
// // * await_ after runned as async // * await_ after runned as async
// // */ // */
// auto aa = async_ ( []() { // auto a = async_ ( []() {
// sleep_for(2s); // only for simulate log duration function // sleep_for(2s); // only for simulate log duration function
// cout << "async_ 2" << endl; // cout << "async_ 2" << endl;
// return 5; // return 5;
// }); // });
// cout << await_(aa) << endl; // cout << await_(a) << endl;
// cout << "print after async_ 2" << endl; // cout << "print after async_ 2" << endl;
// /** // /**
@ -290,19 +271,17 @@ int main () {
// ); // );
// } // }
// auto await_all2 = [&] () { // auto await_all = [&] () {
// for (int i=0; i<fut_vec.size(); i++) { // for (int i=0; i<fut_vec.size(); i++) {
// await_ (fut_vec[i]); // await_ (fut_vec[i]);
// } // }
// }; // };
// await_all2(); // --------------- EVENTS -------------------
// // --------------- EVENTS -------------------
// /** /**
// * initialization of typed events * initialization of typed events
// */ */
// trigger<int, int> ev2int; // trigger<int, int> ev2int;
// trigger<int, string> evintString; // trigger<int, string> evintString;

Loading…
Cancel
Save