AsyncLoop (ThreadPool, task execute loop), interval, timeout, asynco, wait

9 months ago · 03b11840f4
commit 03b11840f4
10 changed files with 399 additions and 0 deletions
--- a/.vscode/c_cpp_properties.json
+++ b/.vscode/c_cpp_properties.json
@ -0,0 +1,16 @@
+{
+    "configurations": [
+        {
+            "name": "Linux",
+            "includePath": [
+                "${workspaceFolder}/**"
+            ],
+            "defines": [],
+            "compilerPath": "/usr/bin/gcc",
+            "cStandard": "c17",
+            "cppStandard": "gnu++17",
+            "intelliSenseMode": "linux-gcc-x64"
+        }
+    ],
+    "version": 4
+}
--- a/.vscode/settings.json
+++ b/.vscode/settings.json
@ -0,0 +1,27 @@
+{
+    "files.associations": {
+        "iostream": "cpp",
+        "functional": "cpp",
+        "thread": "cpp",
+        "chrono": "cpp",
+        "ostream": "cpp",
+        "condition_variable": "cpp",
+        "array": "cpp",
+        "atomic": "cpp",
+        "cwchar": "cpp",
+        "deque": "cpp",
+        "unordered_map": "cpp",
+        "vector": "cpp",
+        "exception": "cpp",
+        "initializer_list": "cpp",
+        "iosfwd": "cpp",
+        "mutex": "cpp",
+        "new": "cpp",
+        "ratio": "cpp",
+        "stdexcept": "cpp",
+        "tuple": "cpp",
+        "type_traits": "cpp",
+        "utility": "cpp",
+        "future": "cpp"
+    }
+}
--- a/.vscode/tasks.json
+++ b/.vscode/tasks.json
@ -0,0 +1,28 @@
+{
+    "tasks": [
+        {
+            "type": "cppbuild",
+            "label": "C/C++: gcc build active file",
+            "command": "/usr/bin/g++",
+            "args": [
+                "-fdiagnostics-color=always",
+                "-g",
+                "${file}",
+                "-o",
+                "${fileDirname}/${fileBasenameNoExtension}"
+            ],
+            "options": {
+                "cwd": "${fileDirname}"
+            },
+            "problemMatcher": [
+                "$gcc"
+            ],
+            "group": {
+                "kind": "build",
+                "isDefault": true
+            },
+            "detail": "Task generated by Debugger."
+        }
+    ],
+    "version": "2.0.0"
+}
--- a/README.md
+++ b/README.md
--- a/lib/asynco.hpp
+++ b/lib/asynco.hpp
@ -0,0 +1,93 @@
+#ifndef _ASYNCO_
+#define _ASYNCO_
+
+#include "loop.hpp"
+
+using namespace std;
+
+namespace marcelb {
+
+#ifndef ON_ASYNC
+#define ON_ASYNC
+AsyncLoop on_async;
+#endif
+
+class interval {
+    bool run = true;
+    function<void()> callback;
+    const time_t _duration;
+    
+    public:
+    interval(function<void()> func, const time_t duration): callback(func), _duration(duration) {
+        #ifndef ON_ASYNC
+        throw string("Not on_async defined!");
+        #endif
+
+        auto task = [&] () {
+            while (run) {
+                this_thread::sleep_for(chrono::milliseconds(_duration));
+                if (run) {
+                    callback();
+                }
+            }
+        };
+
+        on_async.put_task(task);
+    }
+
+    void clear() {
+        run = false;
+    }
+
+    ~interval() {
+        clear();
+    }
+};
+
+class timeout {
+    bool run = true;
+    function<void()> callback;
+    const time_t _duration;
+    
+    public:
+    timeout(function<void()> f, const time_t duration): callback(f), _duration(duration) {
+        #ifndef ON_ASYNC
+        throw string("Not on_async defined!");
+        #endif
+
+        auto task = [&] () {
+            this_thread::sleep_for(chrono::milliseconds(_duration));
+            if (run) {
+                callback();
+            }
+        };
+
+        on_async.put_task(task);
+    }
+
+    void clear() {
+        run = false;
+    }
+
+    ~timeout() {
+        clear();
+    }
+};
+
+
+template<class F, class... Args>
+auto asynco(F&& f, Args&&... args) -> future<typename result_of<F(Args...)>::type> {
+    using return_type = typename result_of<F(Args...)>::type;
+
+    future<return_type> res = on_async.put_task(bind(forward<F>(f), forward<Args>(args)...));
+    return res;
+}
+
+template<typename T>
+T wait(future<T> r) {
+    return r.get();
+}
+
+}
+
+#endif
--- a/lib/event.hpp
+++ b/lib/event.hpp
@ -0,0 +1,40 @@
+#ifndef _EVENT_
+#define _EVENT_
+
+#include <iostream>
+#include <map>
+#include <string>
+#include <functional>
+
+using namespace std;
+
+namespace marcelb {
+
+class event {
+    map<string, function<void(const tuple<>&)>> events;
+public:
+
+    template <typename... Args>
+    void on(const string& key, function<void(Args...)> f) {
+        // events[key] = [f](Args... args) {
+        //     f(args...);
+        // };
+    }
+
+    template <typename... Args>
+    void emit(const string& key, Args&&... args) {
+        if (events.find(key) == events.end()) {
+            cout << "No defined listener for event: " << key << endl;
+            return;
+        }
+        else {
+            for (auto& func : events[key]) {
+                func(forward<Args>(args)...);
+            }
+        }
+    }
+};
+
+}
+
+#endif
--- a/lib/loop.hpp
+++ b/lib/loop.hpp
@ -0,0 +1,91 @@
+#ifndef _LOOP_
+#define _LOOP_
+
+#include <thread>
+#include <vector>
+#include <queue>
+#include <functional>
+#include <mutex>
+#include <condition_variable>
+#include <future>
+
+using namespace std;
+
+namespace marcelb {
+
+#ifdef ON_ASYNC
+extern AsyncLoop on_async;
+#endif
+
+class AsyncLoop {
+    private:
+    vector<thread> workers;
+    queue<function<void()>> tasks;
+    mutex q_io;
+    condition_variable cv;
+    bool stop;
+
+    public:
+    AsyncLoop(size_t pool_size = thread::hardware_concurrency()) : stop(false) {
+        for (size_t i = 0; i < pool_size; ++i) {
+            workers.emplace_back([this] {
+                while (true) {
+                    function<void()> task;
+                    {
+                        unique_lock<mutex> lock(q_io);
+                        cv.wait(lock, [this] { return stop || !tasks.empty(); });
+                        if (stop && tasks.empty())
+                            return;
+                        task = move(tasks.front());
+                        tasks.pop();
+                    }
+                    task();
+                }
+            });
+        }
+    }
+
+    template<class F, class... Args>
+    auto put_task(F&& f, Args&&... args)
+        -> future<typename result_of<F(Args...)>::type> {
+        using return_type = typename result_of<F(Args...)>::type;
+
+        auto task = make_shared<packaged_task<return_type()>>(bind(forward<F>(f), forward<Args>(args)...));
+        future<return_type> res = task->get_future();
+        {
+            unique_lock<mutex> lock(q_io);
+
+            if (stop) {
+                throw runtime_error("Pool is stoped!");
+            }
+
+            tasks.emplace([task]() { (*task)(); });
+        }
+
+        cv.notify_one();
+        return res;
+    }
+
+    unsigned int count_tasks() {
+        return tasks.size();
+    }
+
+    unsigned int count_threads() {
+        return workers.size();
+    }
+
+    ~AsyncLoop() {
+        {
+            unique_lock<mutex> lock(q_io);
+            stop = true;
+        }
+        cv.notify_all();
+        for (thread& worker : workers)
+            worker.join();
+    }
+
+};
+
+}
+
+#endif
--- a/src/pool.cpp
+++ b/src/pool.cpp
--- a/test/test
+++ b/test/test
--- a/test/test.cpp
+++ b/test/test.cpp
@ -0,0 +1,104 @@
+
+#include "../lib/loop.hpp"
+#include "../lib/asynco.hpp"
+// #include "../lib/event.hpp"
+#include <iostream>
+#include <unistd.h>
+
+using namespace std;
+using namespace marcelb;
+using namespace this_thread;
+
+#ifndef ON_ASYNC
+#define ON_ASYNC
+AsyncLoop on_async;
+#endif
+
+int main () {
+
+    auto start = chrono::high_resolution_clock::now();
+
+    interval inter1 ([&]() {
+        cout << "interval prvi " << chrono::duration_cast<chrono::milliseconds>(chrono::high_resolution_clock::now() - start).count() << endl;
+    }, 1000);
+
+    cout << "Blokira stoka" << endl;
+
+    interval inter2 ([&]() {
+        cout << "interval drugi " << chrono::duration_cast<chrono::milliseconds>(chrono::high_resolution_clock::now() - start).count() << endl;
+    }, 2000);
+
+    interval inter3 ([&]() {
+        cout << "interval treći " << chrono::duration_cast<chrono::milliseconds>(chrono::high_resolution_clock::now() - start).count() << endl;
+    }, 3000);
+
+    timeout time1 ( [&] () {
+        cout << "Close interval 1 i 2 " << chrono::duration_cast<chrono::milliseconds>(chrono::high_resolution_clock::now() - start).count() << endl;
+        inter1.clear();
+        inter2.clear();
+    }, 10000);
+
+    timeout time2 ([&] () {
+        cout << "Close interval 3 " << chrono::duration_cast<chrono::milliseconds>(chrono::high_resolution_clock::now() - start).count() << endl;
+        inter3.clear();
+    }, 20000);
+
+
+    //  for (int i = 0; i < 8; ++i) {
+    //     pool.put_task( [&] (int id) {
+    //         this_thread::sleep_for(chrono::seconds(1));
+    //         cout << a*i << endl;
+    //     }, i);
+    // }
+
+    // auto res1 = pool.put_task( [] () {
+    //     cout << "Jebiga " <<endl;
+    //     throw string ("jebiga!!");
+    // });
+
+    // try {
+    //     res1.get();
+    // } catch (const string except) {
+    //     cout << except << endl;
+    // }
+
+    // auto res = pool.put_task( []() {
+    //     this_thread::sleep_for(chrono::seconds(1));
+    //     return 4;
+    // });
+
+    // cout << wait(asynco( [] () {
+    //     sleep_for(chrono::seconds(1));
+    //     cout << "RETURN" << endl;
+    //     return 4;
+    // })) << endl;
+
+    // asynco( []() {
+    //     sleep_for(2s);
+    //     cout << "RETURN 2" << endl;
+    //     return 5;
+    // });
+
+    // cout << wait(pool.put_task( []() {
+    //     this_thread::sleep_for(chrono::seconds(1));
+    //     return 7;
+    // })); 
+
+
+    // cout << wait(run1) << endl;
+
+    // event dog;
+
+    // dog.on("roge", [](int a, int b) {
+    //     cout << "Rogeee" << a << b << endl;
+    // });
+
+    // sleep(5);
+
+    // dog.emit("roge", 5, 8);
+
+    sleep(1000);
+
+    return 0;
+}
+