Support move-only closures in TaskQueue

it allows to create less boil-plate code when marshaling call
with move-only parameters (e.g. unique_ptr) to TaskQueue

Bug: None
Change-Id: I97ddf4f8409af2f83d69fd33267e9a87fb60d901
Reviewed-on: https://webrtc-review.googlesource.com/7619
Commit-Queue: Danil Chapovalov <danilchap@webrtc.org>
Reviewed-by: Karl Wiberg <kwiberg@webrtc.org>
Reviewed-by: Per Kjellander <perkj@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#20271}

rtc_base/task_queue.h

@@ -15,9 +15,11 @@
 #include <memory>
 #include <queue>
 #include <type_traits>
+#include <utility>
 
 #include "rtc_base/constructormagic.h"
 #include "rtc_base/criticalsection.h"
+#include "rtc_base/ptr_util.h"
 #include "rtc_base/scoped_ref_ptr.h"
 
 namespace rtc {
@@ -45,7 +47,8 @@ class QueuedTask {
 template <class Closure>
 class ClosureTask : public QueuedTask {
  public:
-  explicit ClosureTask(const Closure& closure) : closure_(closure) {}
+  explicit ClosureTask(Closure&& closure)
+      : closure_(std::forward<Closure>(closure)) {}
 
  private:
   bool Run() override {
@@ -53,7 +56,8 @@ class ClosureTask : public QueuedTask {
     return true;
   }
 
-  Closure closure_;
+  typename std::remove_const<
+      typename std::remove_reference<Closure>::type>::type closure_;
 };
 
 // Extends ClosureTask to also allow specifying cleanup code.
@@ -62,27 +66,29 @@ class ClosureTask : public QueuedTask {
 template <class Closure, class Cleanup>
 class ClosureTaskWithCleanup : public ClosureTask<Closure> {
  public:
-  ClosureTaskWithCleanup(const Closure& closure, Cleanup cleanup)
-      : ClosureTask<Closure>(closure), cleanup_(cleanup) {}
+  ClosureTaskWithCleanup(Closure&& closure, Cleanup&& cleanup)
+      : ClosureTask<Closure>(std::forward<Closure>(closure)),
+        cleanup_(std::forward<Cleanup>(cleanup)) {}
   ~ClosureTaskWithCleanup() { cleanup_(); }
 
  private:
-  Cleanup cleanup_;
+  typename std::remove_const<
+      typename std::remove_reference<Cleanup>::type>::type cleanup_;
 };
 
 // Convenience function to construct closures that can be passed directly
 // to methods that support std::unique_ptr<QueuedTask> but not template
 // based parameters.
 template <class Closure>
-static std::unique_ptr<QueuedTask> NewClosure(const Closure& closure) {
-  return std::unique_ptr<QueuedTask>(new ClosureTask<Closure>(closure));
+static std::unique_ptr<QueuedTask> NewClosure(Closure&& closure) {
+  return rtc::MakeUnique<ClosureTask<Closure>>(std::forward<Closure>(closure));
 }
 
 template <class Closure, class Cleanup>
-static std::unique_ptr<QueuedTask> NewClosure(const Closure& closure,
-                                              const Cleanup& cleanup) {
-  return std::unique_ptr<QueuedTask>(
-      new ClosureTaskWithCleanup<Closure, Cleanup>(closure, cleanup));
+static std::unique_ptr<QueuedTask> NewClosure(Closure&& closure,
+                                              Cleanup&& cleanup) {
+  return rtc::MakeUnique<ClosureTaskWithCleanup<Closure, Cleanup>>(
+      std::forward<Closure>(closure), std::forward<Cleanup>(cleanup));
 }
 
 // Implements a task queue that asynchronously executes tasks in a way that
@@ -185,50 +191,44 @@ class RTC_LOCKABLE TaskQueue {
   // std::unique_ptr<SomeClassDerivedFromQueuedTask> would not end up being
   // caught by this template.
   template <class Closure,
-            typename std::enable_if<
-                std::is_copy_constructible<Closure>::value>::type* = nullptr>
-  void PostTask(const Closure& closure) {
-    PostTask(std::unique_ptr<QueuedTask>(new ClosureTask<Closure>(closure)));
+            typename std::enable_if<!std::is_convertible<
+                Closure,
+                std::unique_ptr<QueuedTask>>::value>::type* = nullptr>
+  void PostTask(Closure&& closure) {
+    PostTask(NewClosure(std::forward<Closure>(closure)));
   }
 
   // See documentation above for performance expectations.
-  template <class Closure>
-  void PostDelayedTask(const Closure& closure, uint32_t milliseconds) {
-    PostDelayedTask(
-        std::unique_ptr<QueuedTask>(new ClosureTask<Closure>(closure)),
-        milliseconds);
+  template <class Closure,
+            typename std::enable_if<!std::is_convertible<
+                Closure,
+                std::unique_ptr<QueuedTask>>::value>::type* = nullptr>
+  void PostDelayedTask(Closure&& closure, uint32_t milliseconds) {
+    PostDelayedTask(NewClosure(std::forward<Closure>(closure)), milliseconds);
   }
 
   template <class Closure1, class Closure2>
-  void PostTaskAndReply(const Closure1& task,
-                        const Closure2& reply,
+  void PostTaskAndReply(Closure1&& task,
+                        Closure2&& reply,
                         TaskQueue* reply_queue) {
-    PostTaskAndReply(
-        std::unique_ptr<QueuedTask>(new ClosureTask<Closure1>(task)),
-        std::unique_ptr<QueuedTask>(new ClosureTask<Closure2>(reply)),
-        reply_queue);
+    PostTaskAndReply(NewClosure(std::forward<Closure1>(task)),
+                     NewClosure(std::forward<Closure2>(reply)), reply_queue);
   }
 
   template <class Closure>
-  void PostTaskAndReply(std::unique_ptr<QueuedTask> task,
-                        const Closure& reply) {
-    PostTaskAndReply(std::move(task), std::unique_ptr<QueuedTask>(
-                                          new ClosureTask<Closure>(reply)));
+  void PostTaskAndReply(std::unique_ptr<QueuedTask> task, Closure&& reply) {
+    PostTaskAndReply(std::move(task), NewClosure(std::forward<Closure>(reply)));
   }
 
   template <class Closure>
-  void PostTaskAndReply(const Closure& task,
-                        std::unique_ptr<QueuedTask> reply) {
-    PostTaskAndReply(
-        std::unique_ptr<QueuedTask>(new ClosureTask<Closure>(task)),
-        std::move(reply));
+  void PostTaskAndReply(Closure&& task, std::unique_ptr<QueuedTask> reply) {
+    PostTaskAndReply(NewClosure(std::forward<Closure>(task)), std::move(reply));
   }
 
   template <class Closure1, class Closure2>
-  void PostTaskAndReply(const Closure1& task, const Closure2& reply) {
-    PostTaskAndReply(
-        std::unique_ptr<QueuedTask>(new ClosureTask<Closure1>(task)),
-        std::unique_ptr<QueuedTask>(new ClosureTask<Closure2>(reply)));
+  void PostTaskAndReply(Closure1&& task, Closure2&& reply) {
+    PostTaskAndReply(NewClosure(std::forward(task)),
+                     NewClosure(std::forward(reply)));
   }
 
  private:

rtc_base/task_queue_unittest.cc

@@ -239,7 +239,7 @@ TEST(TaskQueueTest, PostAndReuse) {
     Event* const event_;
   };
 
-  std::unique_ptr<QueuedTask> task(
+  std::unique_ptr<ReusedTask> task(
       new ReusedTask(&call_count, &reply_queue, &event));
 
   post_queue.PostTask(std::move(task));
@@ -260,6 +260,105 @@ TEST(TaskQueueTest, PostAndReplyLambda) {
   EXPECT_TRUE(my_flag);
 }
 
+TEST(TaskQueueTest, PostCopyableClosure) {
+  struct CopyableClosure {
+    CopyableClosure(int* num_copies, int* num_moves, Event* event)
+        : num_copies(num_copies), num_moves(num_moves), event(event) {}
+    CopyableClosure(const CopyableClosure& other)
+        : num_copies(other.num_copies),
+          num_moves(other.num_moves),
+          event(other.event) {
+      ++*num_copies;
+    }
+    CopyableClosure(CopyableClosure&& other)
+        : num_copies(other.num_copies),
+          num_moves(other.num_moves),
+          event(other.event) {
+      ++*num_moves;
+    }
+    void operator()() { event->Set(); }
+
+    int* num_copies;
+    int* num_moves;
+    Event* event;
+  };
+
+  int num_copies = 0;
+  int num_moves = 0;
+  Event event(false, false);
+
+  static const char kPostQueue[] = "PostCopyableClosure";
+  TaskQueue post_queue(kPostQueue);
+  {
+    CopyableClosure closure(&num_copies, &num_moves, &event);
+    post_queue.PostTask(closure);
+    // Destroy closure to check with msan and tsan posted task has own copy.
+  }
+
+  EXPECT_TRUE(event.Wait(1000));
+  EXPECT_EQ(num_copies, 1);
+  EXPECT_EQ(num_moves, 0);
+}
+
+TEST(TaskQueueTest, PostMoveOnlyClosure) {
+  struct SomeState {
+    explicit SomeState(Event* event) : event(event) {}
+    ~SomeState() { event->Set(); }
+    Event* event;
+  };
+  struct MoveOnlyClosure {
+    MoveOnlyClosure(int* num_moves, std::unique_ptr<SomeState> state)
+        : num_moves(num_moves), state(std::move(state)) {}
+    MoveOnlyClosure(const MoveOnlyClosure&) = delete;
+    MoveOnlyClosure(MoveOnlyClosure&& other)
+        : num_moves(other.num_moves), state(std::move(other.state)) {
+      ++*num_moves;
+    }
+    void operator()() { state.reset(); }
+
+    int* num_moves;
+    std::unique_ptr<SomeState> state;
+  };
+
+  int num_moves = 0;
+  Event event(false, false);
+  std::unique_ptr<SomeState> state(new SomeState(&event));
+
+  static const char kPostQueue[] = "PostMoveOnlyClosure";
+  TaskQueue post_queue(kPostQueue);
+  post_queue.PostTask(MoveOnlyClosure(&num_moves, std::move(state)));
+
+  EXPECT_TRUE(event.Wait(1000));
+  EXPECT_EQ(num_moves, 1);
+}
+
+TEST(TaskQueueTest, PostMoveOnlyCleanup) {
+  struct SomeState {
+    explicit SomeState(Event* event) : event(event) {}
+    ~SomeState() { event->Set(); }
+    Event* event;
+  };
+  struct MoveOnlyClosure {
+    void operator()() { state.reset(); }
+
+    std::unique_ptr<SomeState> state;
+  };
+
+  Event event_run(false, false);
+  Event event_cleanup(false, false);
+  std::unique_ptr<SomeState> state_run(new SomeState(&event_run));
+  std::unique_ptr<SomeState> state_cleanup(new SomeState(&event_cleanup));
+
+  static const char kPostQueue[] = "PostMoveOnlyCleanup";
+  TaskQueue post_queue(kPostQueue);
+  post_queue.PostTask(NewClosure(MoveOnlyClosure{std::move(state_run)},
+                                 MoveOnlyClosure{std::move(state_cleanup)}));
+
+  EXPECT_TRUE(event_cleanup.Wait(1000));
+  // Expect run closure to complete before cleanup closure.
+  EXPECT_TRUE(event_run.Wait(0));
+}
+
 // This test covers a particular bug that we had in the libevent implementation
 // where we could hit a deadlock while trying to post a reply task to a queue
 // that was being deleted.  The test isn't guaranteed to hit that case but it's

rtc_base/task_queue_win.cc

@@ -15,6 +15,7 @@
 
 #include <algorithm>
 #include <queue>
+#include <utility>
 
 #include "rtc_base/arraysize.h"
 #include "rtc_base/checks.h"
@@ -170,10 +171,11 @@ class TaskQueue::Impl : public RefCountInterface {
   bool IsCurrent() const;
 
   template <class Closure,
-            typename std::enable_if<
-                std::is_copy_constructible<Closure>::value>::type* = nullptr>
-  void PostTask(const Closure& closure) {
-    PostTask(std::unique_ptr<QueuedTask>(new ClosureTask<Closure>(closure)));
+            typename std::enable_if<!std::is_convertible<
+                Closure,
+                std::unique_ptr<QueuedTask>>::value>::type* = nullptr>
+  void PostTask(Closure&& closure) {
+    PostTask(NewClosure(std::forward<Closure>(closure)));
   }
 
   void PostTask(std::unique_ptr<QueuedTask> task);