feat update

CMakeLists.txt

@@ -120,6 +120,7 @@ if(SLED_BUILD_TESTS)
 		src/exec/just_test.cc
 		src/any_test.cc
         src/filesystem/path_test.cc
+		src/futures/future_test.cc
 		# src/profiling/profiling_test.cc
         src/strings/base64_test.cc
         src/cleanup_test.cc

include/sled/exec/detail/just.h

@@ -12,7 +12,7 @@ struct JustOperation {
     TReceiver receiver;
     T value;
 
-    void Start() { receiver.SetValue(std::move(value)); }
+    void Start() { receiver.SetValue(value); }
 };
 
 template<typename T>

include/sled/exec/detail/retry.h

@@ -0,0 +1,67 @@
+#ifndef SLED_EXEC_DETAIL_RETRY_H
+#define SLED_EXEC_DETAIL_RETRY_H
+#include <atomic>
+#include <memory>
+#include <system_error>
+#pragma once
+
+#include "traits.h"
+
+namespace sled {
+
+struct RetryState {
+    int retry_count;
+    bool need_retry;
+};
+
+template<typename TReceiver>
+struct RetryReceiver {
+    TReceiver receiver;
+    std::shared_ptr<RetryState> state;
+
+    template<typename T>
+    void SetValue(T &&value)
+    {
+        receiver.SetValue(value);
+    }
+
+    void SetError(std::exception_ptr err)
+    {
+        if (state->retry_count < 0) {}
+    }
+
+    void SetStopped() { receiver.SetStopped(); }
+};
+
+template<typename TSender, typename TReceiver>
+struct RetryOperation {
+    ConnectResultT<TSender, RetryReceiver<TReceiver>> op;
+    std::shared_ptr<int> state;
+
+    void Start() {}
+};
+
+template<typename TSender>
+struct RetrySender {
+    using S = typename std::remove_cv<typename std::remove_reference<TSender>::type>::type;
+    using result_t = SenderResultT<S>;
+    S sender;
+    int retry_count;
+
+    template<typename TReceiver>
+    RetryOperation<TSender, TReceiver> Connect(TReceiver &&receiver)
+    {
+        auto retry_state = std::make_shared<RetryState>(new RetryState{retry_count, false});
+        return {sender.Connect(RetryReceiver<TReceiver>{receiver, retry_state}), retry_state};
+    }
+};
+
+template<typename TSender>
+RetrySender<TSender>
+Retry(TSender &&sender, int retry_count)
+{
+    return {std::forward<TSender>(sender), retry_count};
+}
+
+}// namespace sled
+#endif//  SLED_EXEC_DETAIL_RETRY_H

include/sled/exec/detail/sync_wait.h

@@ -8,7 +8,7 @@
 #include <exception>
 
 namespace sled {
-struct SyncWaitData {
+struct SyncWaitState {
     sled::Mutex lock;
     sled::ConditionVariable cv;
     std::exception_ptr err;
@@ -17,7 +17,7 @@ struct SyncWaitData {
 
 template<typename T>
 struct SyncWaitReceiver {
-    SyncWaitData &data;
+    SyncWaitState &data;
     sled::optional<T> &value;
 
     void SetValue(T &&val)
@@ -49,7 +49,7 @@ sled::optional<SenderResultT<TSender>>
 SyncWait(TSender sender)
 {
     using T = SenderResultT<TSender>;
-    SyncWaitData data;
+    SyncWaitState data;
     sled::optional<T> value;
 
     auto op = sender.Connect(SyncWaitReceiver<T>{data, value});

include/sled/exec/detail/then.h

@@ -15,7 +15,11 @@ struct ThenReceiver {
     template<typename T>
     void SetValue(T &&value)
     {
+        try {
             receiver.SetValue(func(std::forward<T>(value)));
+        } catch (...) {
+            receiver.SetError(std::current_exception());
+        }
     }
 
     void SetError(std::exception_ptr err) { receiver.SetError(err); }
@@ -32,8 +36,9 @@ struct ThenOperation {
 
 template<typename TSender, typename F>
 struct ThenSender {
-    using result_t = typename eggs::invoke_result_t<F, SenderResultT<TSender>>;
+    using S = typename std::remove_cv<typename std::remove_reference<TSender>::type>::type;
-    TSender sender;
+    using result_t = typename eggs::invoke_result_t<F, SenderResultT<S>>;
+    S sender;
     F func;
 
     template<typename TReceiver>
@@ -45,7 +50,7 @@ struct ThenSender {
 
 template<typename TSender, typename F>
 ThenSender<TSender, F>
-Then(TSender sender, F &&func)
+Then(TSender &&sender, F &&func)
 {
     return {std::forward<TSender>(sender), std::forward<F>(func)};
 }

include/sled/futures/base_cell.h

@@ -0,0 +1,15 @@
+#ifndef SLED_FUTURES_BASE_CELL_H
+#define SLED_FUTURES_BASE_CELL_H
+#include <functional>
+#pragma once
+
+namespace sled {
+namespace futures {
+
+struct BaseCell {
+    void *scheduler;
+};
+}// namespace futures
+
+}// namespace sled
+#endif//  SLED_FUTURES_BASE_CELL_H

include/sled/futures/future.h

@@ -0,0 +1,51 @@
+#ifndef SLED_FUTURES_FUTHRE_H
+#define SLED_FUTURES_FUTHRE_H
+
+#include "sled/any.h"
+#include "sled/exec/detail/invoke_result.h"
+#include "sled/optional.h"
+#include "sled/synchronization/mutex.h"
+#include <exception>
+#include <memory>
+
+namespace sled {
+
+template<typename T>
+class Future;
+template<typename T>
+class Promise;
+
+template<typename T>
+struct FPState : std::enable_shared_from_this<FPState<T>> {
+    sled::Mutex lock;
+    sled::optional<T> data;
+    std::exception_ptr err;
+    bool done;
+    sled::any priv;
+};
+
+template<typename T>
+class Future {
+public:
+    using result_t = T;
+
+    Future(std::shared_ptr<FPState<T>> state) : state_(state) {}
+
+private:
+    std::shared_ptr<FPState<T>> state_;
+};
+
+template<typename T>
+class Promise {
+public:
+    using result_t = T;
+
+    void SetValue(T &&value) {}
+
+    void SetError(std::exception_ptr err) {}
+
+    Future<T> GetFuture() {}
+};
+
+}// namespace sled
+#endif//  SLED_FUTURES_FUTHRE_H

include/sled/futures/just.h

@@ -0,0 +1,28 @@
+#ifndef SLED_FUTURES_JUST_H
+#define SLED_FUTURES_JUST_H
+#include <utility>
+#pragma once
+
+namespace sled {
+namespace futures {
+template<typename T>
+struct JustCell {
+    T value;
+
+    template<typename R>
+    void Start(R receiver)
+    {
+        receiver.SetValue(value);
+    }
+};
+
+template<typename T>
+JustCell<T>
+Just(T &&t)
+{
+    return {std::forward<T>(t)};
+}
+
+}// namespace futures
+}// namespace sled
+#endif//  SLED_FUTURES_JUST_H

include/sled/futures/then.h

@@ -0,0 +1,40 @@
+
+#ifndef SLED_FUTURES_THEN_H
+#define SLED_FUTURES_THEN_H
+#include <exception>
+#include <utility>
+#pragma once
+
+namespace sled {
+namespace futures {
+
+template<typename S, typename F>
+struct ThenCell {
+    S sender;
+    F func;
+
+    // T value;
+
+    template<typename R>
+    void Start(R receiver)
+    {
+        sender.Start();
+    }
+
+    template<typename U>
+    void SetValue(U &&value)
+    {}
+
+    void SetError(std::exception_ptr err) {}
+};
+
+template<typename S, typename F>
+ThenCell<S, F>
+Then(S sender, F &&func)
+{
+    return {std::forward<S>(sender), std::forward<F>(func)};
+}
+
+}// namespace futures
+}// namespace sled
+#endif//  SLED_FUTURES_THEN_H

include/sled/system/thread.h

@@ -59,25 +59,6 @@ public:
     Thread(const Thread &) = delete;
     Thread &operator=(const Thread &) = delete;
 
-    void BlockingCall(std::function<void()> functor,
-                      const Location &location = Location::Current())
-    {
-        BlockingCallImpl(functor, location);
-    }
-
-    template<typename Functor,
-             typename ReturnT = typename std::result_of<Functor()>::type,
-             typename = typename std::enable_if<!std::is_void<ReturnT>::value,
-                                                ReturnT>::type>
-    ReturnT BlockingCall(Functor &&functor,
-                         const Location &location = Location::Current())
-    {
-        ReturnT result;
-        BlockingCall([&] { result = std::forward<Functor>(functor)(); },
-                     location);
-        return result;
-    }
-
     static std::unique_ptr<Thread> CreateWithSocketServer();
     static std::unique_ptr<Thread> Create();
     static Thread *Current();
@@ -122,8 +103,7 @@ protected:
         bool operator<(const DelayedMessage &dmsg) const
         {
             return (dmsg.run_time_ms < run_time_ms)
-                || ((dmsg.run_time_ms == run_time_ms)
+                || ((dmsg.run_time_ms == run_time_ms) && (dmsg.message_number < message_number));
-                    && (dmsg.message_number < message_number));
         }
 
         int64_t delay_ms;
@@ -132,15 +112,12 @@ protected:
         mutable std::function<void()> functor;
     };
 
-    void PostTaskImpl(std::function<void()> &&task,
+    void PostTaskImpl(std::function<void()> &&task, const PostTaskTraits &traits, const Location &location) override;
-                      const PostTaskTraits &traits,
-                      const Location &location) override;
     void PostDelayedTaskImpl(std::function<void()> &&task,
                              TimeDelta delay,
                              const PostDelayedTaskTraits &traits,
                              const Location &location) override;
-    virtual void BlockingCallImpl(std::function<void()> functor,
+    void BlockingCallImpl(std::function<void()> &&functor, const Location &location) override;
-                                  const Location &location);
 
     void DoInit();
     void DoDestroy();
@@ -150,8 +127,7 @@ private:
     std::function<void()> Get(int cmsWait);
     void Dispatch(std::function<void()> &&task);
     static void *PreRun(void *pv);
-    bool WrapCurrentWithThreadManager(ThreadManager *thread_manager,
+    bool WrapCurrentWithThreadManager(ThreadManager *thread_manager, bool need_synchronize_access);
-                                      bool need_synchronize_access);
     bool IsRunning();
 
     // for ThreadManager
@@ -171,8 +147,7 @@ private:
     std::unique_ptr<std::thread> thread_;
     bool owned_;
 
-    std::unique_ptr<TaskQueueBase::CurrentTaskQueueSetter>
+    std::unique_ptr<TaskQueueBase::CurrentTaskQueueSetter> task_queue_registration_;
-        task_queue_registration_;
     friend class ThreadManager;
 };
 

include/sled/system/thread_pool.h

@@ -2,11 +2,12 @@
 #ifndef SLED_SYSTEM_THREAD_POOL_H
 #define SLED_SYSTEM_THREAD_POOL_H
 #include "sled/system/fiber/scheduler.h"
+#include "sled/system/thread.h"
 #include <functional>
 #include <future>
 
 namespace sled {
-class ThreadPool final {
+class ThreadPool final : public TaskQueueBase {
 public:
     /**
     * @param num_threads The number of threads to create in the thread pool. If
@@ -18,16 +19,25 @@ public:
     template<typename F, typename... Args>
     auto submit(F &&f, Args &&...args) -> std::future<decltype(f(args...))>
     {
-        std::function<decltype(f(args...))()> func =
+        std::function<decltype(f(args...))()> func = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
-            std::bind(std::forward<F>(f), std::forward<Args>(args)...);
+        auto task_ptr = std::make_shared<std::packaged_task<decltype(f(args...))()>>(func);
-        auto task_ptr =
+        scheduler_->enqueue(marl::Task([task_ptr]() { (*task_ptr)(); }));
-            std::make_shared<std::packaged_task<decltype(f(args...))()>>(func);
-        scheduler->enqueue(marl::Task([task_ptr]() { (*task_ptr)(); }));
         return task_ptr->get_future();
     }
 
+    void Delete() override;
+
+protected:
+    void PostTaskImpl(std::function<void()> &&task, const PostTaskTraits &traits, const Location &location) override;
+
+    void PostDelayedTaskImpl(std::function<void()> &&task,
+                             TimeDelta delay,
+                             const PostDelayedTaskTraits &traits,
+                             const Location &location) override;
+
 private:
-    sled::Scheduler *scheduler;
+    sled::Scheduler *scheduler_;
+    std::unique_ptr<sled::Thread> delayed_thread_;
 };
 
 }// namespace sled

include/sled/task_queue/task_queue_base.h

@@ -22,30 +22,23 @@ public:
     };
 
     struct Deleter {
-        void operator()(TaskQueueBase *task_queue) const
+        void operator()(TaskQueueBase *task_queue) const { task_queue->Delete(); }
-        {
-            task_queue->Delete();
-        }
     };
 
     virtual void Delete() = 0;
 
-    inline void PostTask(std::function<void()> &&task,
+    inline void PostTask(std::function<void()> &&task, const Location &location = Location::Current())
-                         const Location &location = Location::Current())
     {
         PostTaskImpl(std::move(task), PostTaskTraits{}, location);
     }
 
-    inline void PostDelayedTask(std::function<void()> &&task,
+    inline void
-                                TimeDelta delay,
+    PostDelayedTask(std::function<void()> &&task, TimeDelta delay, const Location &location = Location::Current())
-                                const Location &location = Location::Current())
     {
-        PostDelayedTaskImpl(std::move(task), delay, PostDelayedTaskTraits{},
+        PostDelayedTaskImpl(std::move(task), delay, PostDelayedTaskTraits{}, location);
-                            location);
     }
 
-    inline void
+    inline void PostDelayedHighPrecisionTask(std::function<void()> &&task,
-    PostDelayedHighPrecisionTask(std::function<void()> &&task,
                                              TimeDelta delay,
                                              const Location &location = Location::Current())
     {
@@ -53,8 +46,7 @@ public:
         PostDelayedTaskImpl(std::move(task), delay, traits, location);
     }
 
-    inline void
+    inline void PostDelayedTaskWithPrecision(DelayPrecision precision,
-    PostDelayedTaskWithPrecision(DelayPrecision precision,
                                              std::function<void()> &&task,
                                              TimeDelta delay,
                                              const Location &location = Location::Current())
@@ -69,6 +61,21 @@ public:
         }
     }
 
+    void BlockingCall(std::function<void()> functor, const Location &location = Location::Current())
+    {
+        BlockingCallImpl(std::move(functor), location);
+    }
+
+    template<typename Functor,
+             typename ReturnT = typename std::result_of<Functor()>::type,
+             typename = typename std::enable_if<!std::is_void<ReturnT>::value, ReturnT>::type>
+    ReturnT BlockingCall(Functor &&functor, const Location &location = Location::Current())
+    {
+        ReturnT result;
+        BlockingCall([&] { result = std::forward<Functor>(functor)(); }, location);
+        return result;
+    }
+
     static TaskQueueBase *Current();
 
     bool IsCurrent() const { return Current() == this; };
@@ -77,20 +84,17 @@ protected:
     struct PostTaskTraits {};
 
     struct PostDelayedTaskTraits {
-        PostDelayedTaskTraits(bool high_precision = false)
+        PostDelayedTaskTraits(bool high_precision = false) : high_precision(high_precision) {}
-            : high_precision(high_precision)
-        {}
 
         bool high_precision = false;
     };
 
-    virtual void PostTaskImpl(std::function<void()> &&task,
+    virtual void PostTaskImpl(std::function<void()> &&task, const PostTaskTraits &traits, const Location &location) = 0;
-                              const PostTaskTraits &traits,
-                              const Location &location) = 0;
     virtual void PostDelayedTaskImpl(std::function<void()> &&task,
                                      TimeDelta delay,
                                      const PostDelayedTaskTraits &traits,
                                      const Location &location) = 0;
+    virtual void BlockingCallImpl(std::function<void()> &&task, const Location &location);
     virtual ~TaskQueueBase() = default;
 
     class CurrentTaskQueueSetter {
@@ -98,8 +102,7 @@ protected:
         explicit CurrentTaskQueueSetter(TaskQueueBase *task_queue);
         ~CurrentTaskQueueSetter();
         CurrentTaskQueueSetter(const CurrentTaskQueueSetter &) = delete;
-        CurrentTaskQueueSetter &
+        CurrentTaskQueueSetter &operator=(const CurrentTaskQueueSetter &) = delete;
-        operator=(const CurrentTaskQueueSetter &) = delete;
 
     private:
         TaskQueueBase *const previous_;

src/futures/future_test.cc

@@ -0,0 +1,8 @@
+#include <gtest/gtest.h>
+#include <sled/futures/future.h>
+
+TEST(Future, basic)
+{
+    // sled::Future<int> x;
+    // auto res = x.Then([](int) {});
+}

src/system/thread.cc

@@ -45,8 +45,7 @@ void
 ThreadManager::RemoveInternal(Thread *message_queue)
 {
     MutexLock lock(&cirt_);
-    auto iter = std::find(message_queues_.begin(), message_queues_.end(),
+    auto iter = std::find(message_queues_.begin(), message_queues_.end(), message_queue);
-                          message_queue);
     if (iter != message_queues_.end()) { message_queues_.erase(iter); }
 }
 
@@ -96,8 +95,7 @@ ThreadManager::ProcessAllMessageQueueInternal()
     MutexLock lock(&cirt_);
     for (Thread *queue : message_queues_) {
         queues_not_done.fetch_add(1);
-        auto sub =
+        auto sub = MakeCleanup([&queues_not_done] { queues_not_done.fetch_sub(1); });
-            MakeCleanup([&queues_not_done] { queues_not_done.fetch_sub(1); });
         queue->PostDelayedTask([&sub] {}, TimeDelta::Zero());
     }
 
@@ -115,9 +113,7 @@ ThreadManager::SetCurrentThreadInternal(Thread *message_queue)
 
 Thread::Thread(SocketServer *ss) : Thread(ss, /*do_init=*/true) {}
 
-Thread::Thread(std::unique_ptr<SocketServer> ss)
+Thread::Thread(std::unique_ptr<SocketServer> ss) : Thread(std::move(ss), /*do_init=*/true) {}
-    : Thread(std::move(ss), /*do_init=*/true)
-{}
 
 Thread::Thread(SocketServer *ss, bool do_init)
     : delayed_next_num_(0),
@@ -131,11 +127,7 @@ Thread::Thread(SocketServer *ss, bool do_init)
     if (do_init) { DoInit(); }
 }
 
-Thread::Thread(std::unique_ptr<SocketServer> ss, bool do_init)
+Thread::Thread(std::unique_ptr<SocketServer> ss, bool do_init) : Thread(ss.get(), do_init) { own_ss_ = std::move(ss); }
-    : Thread(ss.get(), do_init)
-{
-    own_ss_ = std::move(ss);
-}
 
 Thread::~Thread()
 {
@@ -244,13 +236,10 @@ Thread::Get(int cmsWait)
             cmsNext = cmsDelayNext;
         } else {
             cmsNext = std::max<int64_t>(0, cmsTotal - cmsElapsed);
-            if ((cmsDelayNext != kForever) && (cmsDelayNext < cmsNext)) {
+            if ((cmsDelayNext != kForever) && (cmsDelayNext < cmsNext)) { cmsNext = cmsDelayNext; }
-                cmsNext = cmsDelayNext;
-            }
         }
         {
-            if (!ss_->Wait(cmsNext == kForever ? SocketServer::kForever
+            if (!ss_->Wait(cmsNext == kForever ? SocketServer::kForever : TimeDelta::Millis(cmsNext),
-                                               : TimeDelta::Millis(cmsNext),
                            /*process_io=*/true)) {
                 return nullptr;
             }
@@ -266,9 +255,7 @@ Thread::Get(int cmsWait)
 }
 
 void
-Thread::PostTaskImpl(std::function<void()> &&task,
+Thread::PostTaskImpl(std::function<void()> &&task, const PostTaskTraits &traits, const Location &location)
-                     const PostTaskTraits &traits,
-                     const Location &location)
 {
     if (IsQuitting()) { return; }
     {
@@ -303,8 +290,7 @@ Thread::PostDelayedTaskImpl(std::function<void()> &&task,
 }
 
 void
-Thread::BlockingCallImpl(std::function<void()> functor,
+Thread::BlockingCallImpl(std::function<void()> &&functor, const Location &location)
-                         const Location &location)
 {
     if (IsQuitting()) { return; }
     if (IsCurrent()) {
@@ -373,8 +359,7 @@ Thread::SetName(const std::string &name, const void *obj)
 void
 Thread::EnsureIsCurrentTaskQueue()
 {
-    task_queue_registration_.reset(
+    task_queue_registration_.reset(new TaskQueueBase::CurrentTaskQueueSetter(this));
-        new TaskQueueBase::CurrentTaskQueueSetter(this));
 }
 
 void
@@ -426,8 +411,7 @@ Thread::PreRun(void *pv)
 }
 
 bool
-Thread::WrapCurrentWithThreadManager(ThreadManager *thread_manager,
+Thread::WrapCurrentWithThreadManager(ThreadManager *thread_manager, bool need_synchronize_access)
-                                     bool need_synchronize_access)
 {
     // assert(!IsRunning());
     owned_ = false;
@@ -498,8 +482,7 @@ Thread::Current()
     return thread;
 }
 
-AutoSocketServerThread::AutoSocketServerThread(SocketServer *ss)
+AutoSocketServerThread::AutoSocketServerThread(SocketServer *ss) : Thread(ss, /*do_init=*/false)
-    : Thread(ss, /*do_init=*/false)
 {
     DoInit();
     old_thread_ = ThreadManager::Instance()->CurrentThread();

src/system/thread_pool.cc

@@ -1,15 +1,39 @@
 #include "sled/system/thread_pool.h"
+#include "sled/system/location.h"
+#include "sled/task_queue/task_queue_base.h"
 
 namespace sled {
 ThreadPool::ThreadPool(int num_threads)
 {
-    if (num_threads == -1) {
+    if (num_threads == -1) { num_threads = std::thread::hardware_concurrency(); }
-        num_threads = std::thread::hardware_concurrency();
+    scheduler_ = new sled::Scheduler(sled::Scheduler::Config().setWorkerThreadCount(num_threads));
-    }
-    scheduler = new sled::Scheduler(
-        sled::Scheduler::Config().setWorkerThreadCount(num_threads));
 }
 
-ThreadPool::~ThreadPool() { delete scheduler; }
+ThreadPool::~ThreadPool() { delete scheduler_; }
+
+void
+ThreadPool::Delete()
+{}
+
+void
+ThreadPool::PostTaskImpl(std::function<void()> &&task, const PostTaskTraits &traits, const Location &location)
+{
+    scheduler_->enqueue(marl::Task([task] { task(); }));
+}
+
+void
+ThreadPool::PostDelayedTaskImpl(std::function<void()> &&task,
+                                TimeDelta delay,
+                                const PostDelayedTaskTraits &traits,
+                                const Location &location)
+{
+    if (traits.high_precision) {
+        delayed_thread_->PostDelayedTaskWithPrecision(TaskQueueBase::DelayPrecision::kHigh, std::move(task), delay,
+                                                      location);
+    } else {
+        delayed_thread_->PostDelayedTaskWithPrecision(TaskQueueBase::DelayPrecision::kLow, std::move(task), delay,
+                                                      location);
+    }
+}
 
 }// namespace sled

src/task_queue/task_queue_base.cc

@@ -1,4 +1,5 @@
 #include "sled/task_queue/task_queue_base.h"
+#include "sled/synchronization/event.h"
 
 namespace sled {
 namespace {
@@ -11,12 +12,22 @@ TaskQueueBase::Current()
     return current;
 }
 
-TaskQueueBase::CurrentTaskQueueSetter::CurrentTaskQueueSetter(TaskQueueBase *task_queue)
+TaskQueueBase::CurrentTaskQueueSetter::CurrentTaskQueueSetter(TaskQueueBase *task_queue) : previous_(current)
-    : previous_(current)
 {
     current = task_queue;
 }
 
 TaskQueueBase::CurrentTaskQueueSetter::~CurrentTaskQueueSetter() { current = previous_; }
 
+void
+TaskQueueBase::BlockingCallImpl(std::function<void()> &&functor, const sled::Location &from)
+{
+    Event done;
+    PostTask([functor, &done] {
+        functor();
+        done.Set();
+    });
+    done.Wait(Event::kForever);
+}
+
 }// namespace sled