add getDefaultUncaughtExceptionHandler; add stop() function for simple_task.

Author: xhawk18

add_ons/simple_task/simple_task.hpp

@@ -41,6 +41,7 @@
 #include <atomic>
 #include <condition_variable>
 #include <utility>
+#include <stdexcept>
 #include "promise-cpp/promise.hpp"
 
 
@@ -55,11 +56,13 @@ class Service {
     Tasks  tasks_;
     std::recursive_mutex mutex_;
     std::condition_variable_any cond_;
-    std::atomic<bool> isAutoExit_;
+    std::atomic<bool> isAutoStop_;
+    std::atomic<bool> isStop_;
 
 public:
     Service()
-        : isAutoExit_(true) {
+        : isAutoStop_(true)
+        , isStop_(false) {
     }
 
     // delay for milliseconds
@@ -92,9 +95,9 @@ class Service {
     }
 
     // Set if the io thread will auto exist if no waiting tasks and timers.
-    void setAutoExit(bool isAutoExit) {
+    void setAutoStop(bool isAutoExit) {
         std::lock_guard<std::recursive_mutex> lock(mutex_);
-        isAutoExit_ = isAutoExit;
+        isAutoStop_ = isAutoExit;
         cond_.notify_one();
     }
 
@@ -103,14 +106,14 @@ class Service {
     void run() {
         std::unique_lock<std::recursive_mutex> lock(mutex_);
 
-        while(!isAutoExit_ || tasks_.size() > 0 || timers_.size() > 0){
+        while(!isStop_ && (!isAutoStop_ || tasks_.size() > 0 || timers_.size() > 0)) {
 
             if (tasks_.size() == 0 && timers_.size() == 0) {
                 cond_.wait(lock);
                 continue;
             }
 
-            while (timers_.size() > 0) {
+            while (!isStop_ && timers_.size() > 0) {
                 TimePoint now = std::chrono::steady_clock::now();
                 TimePoint time = timers_.begin()->first;
                 if (time <= now) {
@@ -128,7 +131,7 @@ class Service {
             }
 
             // Check fixed size of tasks in this loop, so that timer have a chance to run.
-            if(tasks_.size() > 0) {
+            if(!isStop_ && tasks_.size() > 0) {
                 size_t size = tasks_.size();
                 for(size_t i = 0; i < size; ++i){
                     Defer defer = tasks_.front();
@@ -137,6 +140,27 @@ class Service {
                 }
             }
         }
+
+        // Clear pending timers and tasks
+        while (timers_.size() > 0 || tasks_.size()) {
+            while (timers_.size() > 0) {
+                Defer defer = timers_.begin()->second;
+                timers_.erase(timers_.begin());
+                defer.reject(std::runtime_error("service stopped"));
+            }
+            while (tasks_.size() > 0) {
+                Defer defer = tasks_.front();
+                tasks_.pop_front();
+                defer.reject(std::runtime_error("service stopped"));
+            }
+        }
+    }
+
+    // stop the service loop
+    void stop() {
+        std::lock_guard<std::recursive_mutex> lock(mutex_);
+        isStop_ = true;
+        cond_.notify_one();
     }
 };
 

example/multithread_test.cpp

@@ -107,7 +107,7 @@ int main() {
             return test_switch(io, 100);
         }).then([&]() {
 
-            io.setAutoExit(false);
+            io.setAutoStop(false);
             // Run in new thread
             return newPromise([](Defer &defer) {
                 std::thread([=]() {
@@ -127,7 +127,7 @@ int main() {
             });
 
         }).then([&]() {
-            io.setAutoExit(true);
+            io.setAutoStop(true);
 
             return test_switch(io, 1000);
         }).then([&, loop]() {

example/simple_benchmark_test.cpp

@@ -94,6 +94,7 @@ int main() {
         test_switch(io, 1).then([&]() {
             return test_switch(io, 1000);
         }).then([&]() {
+            io.stop();
             return test_switch(io, 10000);
         }).then([&]() {
             return test_switch(io, 100000);

include/promise-cpp/promise.hpp

@@ -100,6 +100,7 @@ struct PromiseHolder {
 
     PROMISE_API void dump() const;
     PROMISE_API static any *getUncaughtExceptionHandler();
+    PROMISE_API static any *getDefaultUncaughtExceptionHandler();
     PROMISE_API static void onUncaughtException(const any &arg);
     PROMISE_API static void handleUncaughtException(const any &onUncaughtException);
 };

include/promise-cpp/promise_inl.hpp

@@ -402,24 +402,36 @@ PromiseHolder::~PromiseHolder() {
     }
 }
 
+
+
 any *PromiseHolder::getUncaughtExceptionHandler() {
     static any onUncaughtException;
     return &onUncaughtException;
 }
 
+any *PromiseHolder::getDefaultUncaughtExceptionHandler() {
+    static any defaultUncaughtExceptionHandler = [](Promise &d) {
+        d.fail([](const std::runtime_error &err) {
+            fprintf(stderr, "onUncaughtException in line %d, %s\n", __LINE__, err.what());
+        }).fail([]() {
+            //go here for all other uncaught parameters.
+            fprintf(stderr, "onUncaughtException in line %d\n", __LINE__);
+        });
+    };
+
+    return &defaultUncaughtExceptionHandler;
+}
+
 void PromiseHolder::onUncaughtException(const any &arg) {
     any *onUncaughtException = getUncaughtExceptionHandler();
-    if (onUncaughtException != nullptr && !onUncaughtException->empty()) {
-        try {
-            onUncaughtException->call(reject(arg));
-        }
-        catch (...) {
-            // std::rethrow_exception(std::current_exception());
-            fprintf(stderr, "onUncaughtException in line %d\n", __LINE__);
-        }
+    if (onUncaughtException == nullptr || onUncaughtException->empty()) {
+        onUncaughtException = getDefaultUncaughtExceptionHandler();
     }
-    else {
-        //throw arg;
+
+    try {
+        onUncaughtException->call(reject(arg));
+    }
+    catch (...) {
         fprintf(stderr, "onUncaughtException in line %d\n", __LINE__);
     }
 }