Added dlib::async() and default_thread_pool()

dlib/CMakeLists.txt

@@ -184,6 +184,7 @@ if (NOT TARGET dlib)
          threads/threads_kernel_2.cpp
          threads/threads_kernel_shared.cpp
          threads/thread_pool_extension.cpp
+         threads/async.cpp
          timer/timer.cpp
          stack_trace.cpp
          )

dlib/all/source.cpp

@@ -26,6 +26,7 @@
 #if __cplusplus >= 201103
 #include "../dnn/cpu_dlib.cpp"
 #include "../dnn/tensor_tools.cpp"
+#include "../threads/async.cpp"
 #endif 
 
 #ifndef DLIB_ISO_CPP_ONLY

dlib/test/thread_pool.cpp

@@ -87,11 +87,11 @@ namespace
             add_functor f;
             for (int num_threads= 0; num_threads < 4; ++num_threads)
             {
-                future<int> a, b, c, res, d;
+                dlib::future<int> a, b, c, res, d;
                 thread_pool tp(num_threads);
                 print_spinner();
 
-                future<some_struct> obj;
+                dlib::future<some_struct> obj;
 
 
                 for (int i = 0; i < 4; ++i)

dlib/test/threads.cpp

@@ -19,6 +19,31 @@ namespace
 
     logger dlog("test.threads");
 
+    void test_async()
+    {
+        print_spinner();
+        auto v1 = dlib::async([]() { dlib::sleep(500); return 1; }).share();
+        auto v2 = dlib::async([v1]() { dlib::sleep(400); return v1.get()+1; }).share();
+        auto v3 = dlib::async([v2](int a) { dlib::sleep(300); return v2.get()+a; },2).share();
+        auto v4 = dlib::async([v3]() { dlib::sleep(200); return v3.get()+1; });
+
+        DLIB_TEST(v4.get() == 5);
+
+        print_spinner();
+        auto except = dlib::async([](){ dlib::sleep(300); throw error("oops"); });
+        bool got_exception = false;
+        try
+        {
+            except.get();
+        }
+        catch (error&e)
+        {
+            got_exception = true;
+            DLIB_TEST(e.what() == string("oops"));
+        }
+        DLIB_TEST(got_exception);
+    }
+
     class threads_tester : public tester
     {
     public:
@@ -66,6 +91,8 @@ namespace
 
 
             DLIB_TEST(!failure);
+
+            test_async();
         }
 
         void thread_end_handler (
@@ -118,6 +145,7 @@ namespace
             }
             dlog << LTRACE << "ending of thread num " << num;
 
+
         }
     } a;
 

dlib/threads.h

@@ -23,5 +23,11 @@
 #include "threads/read_write_mutex_extension.h"
 #include "threads/parallel_for_extension.h"
 
+
+// things that require C++11
+#if __cplusplus >= 201103
+#include "threads/async.h"
+#endif
+
 #endif // DLIB_THREADs_
 

dlib/threads/async.cpp

+// Copyright (C) 2016  Davis E. King (davis@dlib.net)
+// License: Boost Software License   See LICENSE.txt for the full license.
+#ifndef DLIB_AsYNC_CPP_
+#define DLIB_AsYNC_CPP_ 
+
+#include "async.h"
+#include <stdlib.h>
+#include "../string.h"
+
+namespace dlib
+{
+
+// ----------------------------------------------------------------------------------------
+
+    namespace impl
+    {
+        unsigned long default_num_threads()
+        {
+            try
+            {
+                char* nt = getenv("DLIB_NUM_THREADS");
+                if (nt)
+                    return string_cast<unsigned long>(nt);
+            } catch(string_cast_error&) {}
+            return std::thread::hardware_concurrency();
+        }
+    }
+
+// ----------------------------------------------------------------------------------------
+
+    thread_pool& default_thread_pool()
+    {
+        static thread_pool tp(impl::default_num_threads());
+        return tp;
+    }
+}
+
+// ----------------------------------------------------------------------------------------
+
+#endif // DLIB_AsYNC_CPP_
+
+

dlib/threads/async.h

+// Copyright (C) 2016  Davis E. King (davis@dlib.net)
+// License: Boost Software License   See LICENSE.txt for the full license.
+#ifndef DLIB_AsYNC_Hh_
+#define DLIB_AsYNC_Hh_ 
+
+#include "async_abstract.h"
+#include "thread_pool_extension.h"
+#include <future>
+#include <functional>
+
+namespace dlib
+{
+
+// ----------------------------------------------------------------------------------------
+
+    namespace impl
+    {
+        template <typename T> struct selector {};
+
+        template <typename T, typename U, typename V>
+        void call_prom_set_value(
+            T& prom,
+            U& fun,
+            selector<V> 
+        )
+        {
+            prom.set_value(fun());
+        }
+
+        template <typename T, typename U>
+        void call_prom_set_value(
+            T& prom,
+            U& fun,
+            selector<void>
+        )
+        {
+            fun();
+            prom.set_value();
+        }
+    }
+
+// ----------------------------------------------------------------------------------------
+
+    thread_pool& default_thread_pool();
+
+// ----------------------------------------------------------------------------------------
+
+    template < 
+        typename Function, 
+        typename ...Args
+        >
+    std::future<typename std::result_of<Function(Args...)>::type> async(
+        thread_pool& tp, 
+        Function&& f, 
+        Args&&... args 
+    )
+    {
+        auto prom = std::make_shared<std::promise<typename std::result_of<Function(Args...)>::type>>();
+        std::future<typename std::result_of<Function(Args...)>::type> ret = prom->get_future();
+        using bind_t = decltype(std::bind(std::forward<Function>(f), std::forward<Args>(args)...));
+        auto fun = std::make_shared<bind_t>(std::bind(std::forward<Function>(f), std::forward<Args>(args)...));
+        tp.add_task_by_value([fun, prom]()
+        { 
+            try
+            {
+                impl::call_prom_set_value(*prom, *fun, impl::selector<typename std::result_of<Function(Args...)>::type>());
+            }
+            catch(...)
+            {
+                prom->set_exception(std::current_exception());
+            }
+        });
+        return std::move(ret);
+    }
+
+// ----------------------------------------------------------------------------------------
+
+    template < 
+        typename Function, 
+        typename ...Args
+        >
+    std::future<typename std::result_of<Function(Args...)>::type> async(
+        Function&& f, 
+        Args&&... args 
+    )
+    {
+        return async(default_thread_pool(), std::forward<Function>(f), std::forward<Args>(args)...);
+    }
+
+}
+
+// ----------------------------------------------------------------------------------------
+
+#ifdef NO_MAKEFILE
+#include "async.cpp"
+#endif
+
+#endif // DLIB_AsYNC_Hh_
+
+
+

dlib/threads/async_abstract.h

+// Copyright (C) 2016  Davis E. King (davis@dlib.net)
+// License: Boost Software License   See LICENSE.txt for the full license.
+#undef DLIB_AsYNC_ABSTRACT_Hh_
+#ifdef DLIB_AsYNC_ABSTRACT_Hh_ 
+
+#include "thread_pool_extension_abstract.h"
+#include <future>
+#include <functional>
+
+namespace dlib
+{
+
+// ----------------------------------------------------------------------------------------
+
+    thread_pool& default_thread_pool(
+    );
+    /*!
+        ensures
+            - returns a reference to a global thread_pool.  If the DLIB_NUM_THREADS
+              environment variable is set to an integer then the thread pool will contain
+              DLIB_NUM_THREADS threads, otherwise it will contain
+              std::thread::hardware_concurrency() threads.
+    !*/
+
+// ----------------------------------------------------------------------------------------
+
+    template < 
+        typename Function, 
+        typename ...Args
+        >
+    std::future<typename std::result_of<Function(Args...)>::type> async(
+        thread_pool& tp, 
+        Function&& f, 
+        Args&&... args 
+    );
+    /*!
+        requires
+            - f must be a function and f(args...) must be a valid expression.
+        ensures
+            - This function behaves just like std::async(std::launch::async, f, args)
+              except that instead of spawning a new thread to process each task it submits
+              the task to the provided dlib::thread_pool.  Therefore, dlib::async() is
+              guaranteed to use a bounded number of threads unlike std::async().  This also
+              means that calls to dlib::async() will block if there aren't any free threads
+              in the thread pool.
+    !*/
+
+// ----------------------------------------------------------------------------------------
+
+    template < 
+        typename Function, 
+        typename ...Args
+        >
+    std::future<typename std::result_of<Function(Args...)>::type> async(
+        Function&& f, 
+        Args&&... args 
+    );
+    /*!
+        ensures
+            - Calling this function is equivalent to directly calling async(default_thread_pool(), f, args...)
+    !*/
+}
+
+// ----------------------------------------------------------------------------------------
+
+#endif // DLIB_AsYNC_ABSTRACT_Hh_
+