Revert "task: stop wrapping tasks with unique_ptr"

This reverts commit 33406cf. It
introduces memory leaks:

Direct leak of 24 byte(s) in 1 object(s) allocated from:
    #0 0x7fb773b389d7 in operator new(unsigned long) (/lib64/libasan.so.5+0x10f9d7)
    #1 0x108f0d4 in seastar::reactor::poller::~poller() ../src/core/reactor.cc:2879
    #2 0x11c1e59 in std::experimental::fundamentals_v1::_Optional_base<seastar::reactor::poller, true>::~_Optional_base() /usr/include/c++/9/experimental/optional:288
    #3 0x118f2d7 in std::experimental::fundamentals_v1::optional<seastar::reactor::poller>::~optional() /usr/include/c++/9/experimental/optional:491
    #4 0x108c5a5 in seastar::reactor::run() ../src/core/reactor.cc:2587
    #5 0xf1a822 in seastar::app_template::run_deprecated(int, char**, std::function<void ()>&&) ../src/core/app-template.cc:199
    #6 0xf1885d in seastar::app_template::run(int, char**, std::function<seastar::future<int> ()>&&) ../src/core/app-template.cc:115
    #7 0xeb2735 in operator() ../src/testing/test_runner.cc:72
    #8 0xebb342 in _M_invoke /usr/include/c++/9/bits/std_function.h:300
    #9 0xf3d8b0 in std::function<void ()>::operator()() const /usr/include/c++/9/bits/std_function.h:690
    #10 0x1034c72 in seastar::posix_thread::start_routine(void*) ../src/core/posix.cc:52
    #11 0x7fb7738804e1 in start_thread /usr/src/debug/glibc-2.30-13-g919af705ee/nptl/pthread_create.c:479

Reported-by: Rafael Avila de Espindola <espindola@scylladb.com>

include/seastar/core/do_with.hh

@@ -90,13 +90,13 @@ public:
 template<typename T, typename F>
 inline
 auto do_with(T&& rvalue, F&& f) {
-    auto task = new internal::do_with_state<T, std::result_of_t<F(T&)>>(std::forward<T>(rvalue));
+    auto task = std::make_unique<internal::do_with_state<T, std::result_of_t<F(T&)>>>(std::forward<T>(rvalue));
     auto fut = f(task->data());
     if (fut.available()) {
         return fut;
     }
     auto ret = task->get_future();
-    internal::set_callback(fut, task);
+    internal::set_callback(fut, std::move(task));
     return ret;
 }
 
@@ -148,13 +148,13 @@ do_with(T1&& rv1, T2&& rv2, T3_or_F&& rv3, More&&... more) {
     auto&& just_func = std::move(std::get<nr>(std::move(all)));
     using value_tuple = std::remove_reference_t<decltype(just_values)>;
     using ret_type = decltype(apply(just_func, just_values));
-    auto task = new internal::do_with_state<value_tuple, ret_type>(std::move(just_values));
+    auto task = std::make_unique<internal::do_with_state<value_tuple, ret_type>>(std::move(just_values));
     auto fut = apply(just_func, task->data());
     if (fut.available()) {
         return fut;
     }
     auto ret = task->get_future();
-    internal::set_callback(fut, task);
+    internal::set_callback(fut, std::move(task));
     return ret;
 }
 

include/seastar/core/future-util.hh

@@ -119,7 +119,7 @@ private:
     // Wait for one of the futures in _incomplete to complete, and then
     // decide what to do: wait for another one, or deliver _result if all
     // are complete.
-    void wait_for_one() noexcept {
+    void wait_for_one() {
         // Process from back to front, on the assumption that the front
         // futures are likely to complete earlier than the back futures.
         // If that's indeed the case, then the front futures will be
@@ -135,7 +135,7 @@ private:
 
         // If there's an incompelete future, wait for it.
         if (!_incomplete.empty()) {
-            internal::set_callback(_incomplete.back(), static_cast<continuation_base<>*>(this));
+            internal::set_callback(_incomplete.back(), std::unique_ptr<continuation_base<>>(this));
             // This future's state will be collected in run_and_dispose(), so we can drop it.
             _incomplete.pop_back();
             return;
@@ -280,14 +280,13 @@ public:
     }
     future<> get_future() { return _promise.get_future(); }
     virtual void run_and_dispose() noexcept override {
+        std::unique_ptr<repeater> zis{this};
         if (_state.failed()) {
             _promise.set_exception(std::move(_state).get_exception());
-            delete this;
             return;
         } else {
             if (std::get<0>(_state.get()) == stop_iteration::yes) {
                 _promise.set_value();
-                delete this;
                 return;
             }
             _state = {};
@@ -296,22 +295,20 @@ public:
             do {
                 auto f = futurator::apply(_action);
                 if (!f.available()) {
-                    internal::set_callback(f, this);
+                    internal::set_callback(f, std::move(zis));
                     return;
                 }
                 if (f.get0() == stop_iteration::yes) {
                     _promise.set_value();
-                    delete this;
                     return;
                 }
             } while (!need_preempt());
         } catch (...) {
             _promise.set_exception(std::current_exception());
-            delete this;
             return;
         }
         _state.set(stop_iteration::no);
-        schedule(this);
+        schedule(std::move(zis));
     }
 };
 
@@ -331,7 +328,7 @@ public:
 template<typename AsyncAction>
 GCC6_CONCEPT( requires seastar::ApplyReturns<AsyncAction, stop_iteration> || seastar::ApplyReturns<AsyncAction, future<stop_iteration>> )
 inline
-future<> repeat(AsyncAction action) noexcept {
+future<> repeat(AsyncAction action) {
     using futurator = futurize<std::result_of_t<AsyncAction()>>;
     static_assert(std::is_same<future<stop_iteration>, typename futurator::type>::value, "bad AsyncAction signature");
     try {
@@ -342,9 +339,9 @@ future<> repeat(AsyncAction action) noexcept {
             if (!f.available()) {
               return [&] () noexcept {
                 memory::disable_failure_guard dfg;
-                auto repeater = new internal::repeater<AsyncAction>(std::move(action));
+                auto repeater = std::make_unique<internal::repeater<AsyncAction>>(std::move(action));
                 auto ret = repeater->get_future();
-                internal::set_callback(f, repeater);
+                internal::set_callback(f, std::move(repeater));
                 return ret;
               }();
             }
@@ -354,9 +351,9 @@ future<> repeat(AsyncAction action) noexcept {
             }
         } while (!need_preempt());
 
-        auto repeater = new internal::repeater<AsyncAction>(stop_iteration::no, std::move(action));
+        auto repeater = std::make_unique<internal::repeater<AsyncAction>>(stop_iteration::no, std::move(action));
         auto ret = repeater->get_future();
-        schedule(repeater);
+        schedule(std::move(repeater));
         return ret;
     } catch (...) {
         return make_exception_future(std::current_exception());
@@ -400,15 +397,14 @@ public:
     }
     future<T> get_future() { return _promise.get_future(); }
     virtual void run_and_dispose() noexcept override {
+        std::unique_ptr<repeat_until_value_state> zis{this};
         if (this->_state.failed()) {
             _promise.set_exception(std::move(this->_state).get_exception());
-            delete this;
             return;
         } else {
             auto v = std::get<0>(std::move(this->_state).get());
             if (v) {
                 _promise.set_value(std::move(*v));
-                delete this;
                 return;
             }
             this->_state = {};
@@ -417,23 +413,21 @@ public:
             do {
                 auto f = futurator::apply(_action);
                 if (!f.available()) {
-                    internal::set_callback(f, this);
+                    internal::set_callback(f, std::move(zis));
                     return;
                 }
                 auto ret = f.get0();
                 if (ret) {
                     _promise.set_value(std::make_tuple(std::move(*ret)));
-                    delete this;
                     return;
                 }
             } while (!need_preempt());
         } catch (...) {
             _promise.set_exception(std::current_exception());
-            delete this;
             return;
         }
         this->_state.set(compat::nullopt);
-        schedule(this);
+        schedule(std::move(zis));
     }
 };
 
@@ -456,7 +450,7 @@ GCC6_CONCEPT( requires requires (AsyncAction aa) {
     futurize<std::result_of_t<AsyncAction()>>::apply(aa).get0().value();
 } )
 repeat_until_value_return_type<AsyncAction>
-repeat_until_value(AsyncAction action) noexcept {
+repeat_until_value(AsyncAction action) {
     using futurator = futurize<std::result_of_t<AsyncAction()>>;
     using type_helper = repeat_until_value_type_helper<typename futurator::type>;
     // the "T" in the documentation
@@ -468,9 +462,9 @@ repeat_until_value(AsyncAction action) noexcept {
         if (!f.available()) {
           return [&] () noexcept {
             memory::disable_failure_guard dfg;
-            auto state = new internal::repeat_until_value_state<AsyncAction, value_type>(std::move(action));
+            auto state = std::make_unique<internal::repeat_until_value_state<AsyncAction, value_type>>(std::move(action));
             auto ret = state->get_future();
-            internal::set_callback(f, state);
+            internal::set_callback(f, std::move(state));
             return ret;
           }();
         }
@@ -486,9 +480,9 @@ repeat_until_value(AsyncAction action) noexcept {
     } while (!need_preempt());
 
     try {
-        auto state = new internal::repeat_until_value_state<AsyncAction, value_type>(compat::nullopt, std::move(action));
+        auto state = std::make_unique<internal::repeat_until_value_state<AsyncAction, value_type>>(compat::nullopt, std::move(action));
         auto f = state->get_future();
-        schedule(state);
+        schedule(std::move(state));
         return f;
     } catch (...) {
         return make_exception_future<value_type>(std::current_exception());
@@ -506,10 +500,10 @@ public:
     explicit do_until_state(StopCondition stop, AsyncAction action) : _stop(std::move(stop)), _action(std::move(action)) {}
     future<> get_future() { return _promise.get_future(); }
     virtual void run_and_dispose() noexcept override {
+        std::unique_ptr<do_until_state> zis{this};
         if (_state.available()) {
             if (_state.failed()) {
                 _promise.set_urgent_state(std::move(_state));
-                delete this;
                 return;
             }
             _state = {}; // allow next cycle to overrun state
@@ -518,26 +512,23 @@ public:
             do {
                 if (_stop()) {
                     _promise.set_value();
-                    delete this;
                     return;
                 }
                 auto f = _action();
                 if (!f.available()) {
-                    internal::set_callback(f, this);
+                    internal::set_callback(f, std::move(zis));
                     return;
                 }
                 if (f.failed()) {
                     f.forward_to(std::move(_promise));
-                    delete this;
                     return;
                 }
             } while (!need_preempt());
         } catch (...) {
             _promise.set_exception(std::current_exception());
-            delete this;
             return;
         }
-        schedule(this);
+        schedule(std::move(zis));
     }
 };
 
@@ -556,7 +547,7 @@ public:
 template<typename AsyncAction, typename StopCondition>
 GCC6_CONCEPT( requires seastar::ApplyReturns<StopCondition, bool> && seastar::ApplyReturns<AsyncAction, future<>> )
 inline
-future<> do_until(StopCondition stop_cond, AsyncAction action) noexcept {
+future<> do_until(StopCondition stop_cond, AsyncAction action) {
     using namespace internal;
     using futurator = futurize<void>;
     do {
@@ -567,9 +558,9 @@ future<> do_until(StopCondition stop_cond, AsyncAction action) noexcept {
         if (!f.available()) {
           return [&] () noexcept {
             memory::disable_failure_guard dfg;
-            auto task = new do_until_state<StopCondition, AsyncAction>(std::move(stop_cond), std::move(action));
+            auto task = std::make_unique<do_until_state<StopCondition, AsyncAction>>(std::move(stop_cond), std::move(action));
             auto ret = task->get_future();
-            internal::set_callback(f, task);
+            internal::set_callback(f, std::move(task));
             return ret;
           }();
         }
@@ -578,9 +569,9 @@ future<> do_until(StopCondition stop_cond, AsyncAction action) noexcept {
         }
     } while (!need_preempt());
 
-    auto task = new do_until_state<StopCondition, AsyncAction>(std::move(stop_cond), std::move(action));
+    auto task = std::make_unique<do_until_state<StopCondition, AsyncAction>>(std::move(stop_cond), std::move(action));
     auto f = task->get_future();
-    schedule(task);
+    schedule(std::move(task));
     return f;
 }
 
@@ -746,7 +737,7 @@ public:
             return true;
         } else {
             auto c = new (continuation) when_all_state_component(wasb, f);
-            set_callback(*f, c);
+            set_callback(*f, std::unique_ptr<when_all_state_component>(c));
             return false;
         }
     }

include/seastar/core/future.hh

@@ -495,7 +495,7 @@ template <typename... T>
 future<T...> make_exception_future(future_state_base&& state) noexcept;
 
 template <typename... T, typename U>
-void set_callback(future<T...>& fut, U* callback) noexcept;
+void set_callback(future<T...>& fut, std::unique_ptr<U> callback);
 
 class future_base;
 
@@ -509,7 +509,7 @@ protected:
     // details.
     future_state_base* _state;
 
-    task* _task = nullptr;
+    std::unique_ptr<task> _task;
 
     promise_base(const promise_base&) = delete;
     promise_base(future_state_base* state) noexcept : _state(state) {}
@@ -607,19 +607,19 @@ public:
 #if SEASTAR_COROUTINES_TS
     void set_coroutine(future_state<T...>& state, task& coroutine) noexcept {
         _state = &state;
-        _task = &coroutine;
+        _task = std::unique_ptr<task>(&coroutine);
     }
 #endif
 private:
     template <typename Func>
-    void schedule(Func&& func) noexcept {
-        auto tws = new continuation<Func, T...>(std::move(func));
+    void schedule(Func&& func) {
+        auto tws = std::make_unique<continuation<Func, T...>>(std::move(func));
         _state = &tws->_state;
-        _task = tws;
+        _task = std::move(tws);
     }
-    void schedule(continuation_base<T...>* callback) noexcept {
+    void schedule(std::unique_ptr<continuation_base<T...>> callback) {
         _state = &callback->_state;
-        _task = callback;
+        _task = std::move(callback);
     }
 
     template <typename... U>
@@ -968,12 +968,12 @@ private:
         return static_cast<internal::promise_base_with_type<T...>*>(future_base::detach_promise());
     }
     template <typename Func>
-    void schedule(Func&& func) noexcept {
+    void schedule(Func&& func) {
         if (_state.available() || !_promise) {
             if (__builtin_expect(!_state.available() && !_promise, false)) {
                 _state.set_to_broken_promise();
             }
-            ::seastar::schedule(new continuation<Func, T...>(std::move(func), std::move(_state)));
+            ::seastar::schedule(std::make_unique<continuation<Func, T...>>(std::move(func), std::move(_state)));
         } else {
             assert(_promise);
             detach_promise()->schedule(std::move(func));
@@ -1095,7 +1095,7 @@ private:
         auto thread = thread_impl::get();
         assert(thread);
         thread_wake_task wake_task{thread, this};
-        detach_promise()->schedule(static_cast<continuation_base<T...>*>(&wake_task));
+        detach_promise()->schedule(std::unique_ptr<continuation_base<T...>>(&wake_task));
         thread_impl::switch_out(thread);
     }
 
@@ -1443,13 +1443,13 @@ public:
     }
 #endif
 private:
-    void set_callback(continuation_base<T...>* callback) noexcept {
+    void set_callback(std::unique_ptr<continuation_base<T...>> callback) {
         if (_state.available()) {
             callback->set_state(get_available_state_ref());
-            ::seastar::schedule(callback);
+            ::seastar::schedule(std::move(callback));
         } else {
             assert(_promise);
-            detach_promise()->schedule(callback);
+            detach_promise()->schedule(std::move(callback));
         }
 
     }
@@ -1470,7 +1470,7 @@ private:
     template <typename... U>
     friend future<U...> internal::make_exception_future(future_state_base&& state) noexcept;
     template <typename... U, typename V>
-    friend void internal::set_callback(future<U...>&, V*) noexcept;
+    friend void internal::set_callback(future<U...>&, std::unique_ptr<V>);
     /// \endcond
 };
 
@@ -1717,10 +1717,10 @@ namespace internal {
 
 template <typename... T, typename U>
 inline
-void set_callback(future<T...>& fut, U* callback) noexcept {
+void set_callback(future<T...>& fut, std::unique_ptr<U> callback) {
     // It would be better to use continuation_base<T...> for U, but
     // then a derived class of continuation_base<T...> won't be matched
-    return fut.set_callback(callback);
+    return fut.set_callback(std::move(callback));
 }
 
 }