Remove unnecessary copies from AsyncInvoke

Currently, the way the AsyncInvoke is implemented, the lambda invoked is copied multiple times. This causes two problems: (1) a reduced performance where captured variables are copied unnecessarily, (2) lambdas with non-copyable captures are not possible to invoke.

This cl attempts to address both points.

Change-Id: I8d907287d6e4851330d469f184760d165fa8bc08
Bug: webrtc:9028
Reviewed-on: https://webrtc-review.googlesource.com/61346
Commit-Queue: Tommi <tommi@webrtc.org>
Reviewed-by: Tommi <tommi@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#22471}

rtc_base/asyncinvoker-inl.h

@@ -49,13 +49,13 @@ template <class FunctorT>
 class FireAndForgetAsyncClosure : public AsyncClosure {
  public:
   explicit FireAndForgetAsyncClosure(AsyncInvoker* invoker,
-                                     const FunctorT& functor)
+                                     FunctorT&& functor)
-      : AsyncClosure(invoker), functor_(functor) {}
+      : AsyncClosure(invoker), functor_(std::forward<FunctorT>(functor)) {}
   virtual void Execute() {
     functor_();
   }
  private:
-  FunctorT functor_;
+  typename std::decay<FunctorT>::type functor_;
 };
 
 }  // namespace rtc

rtc_base/asyncinvoker.h

@@ -97,10 +97,11 @@ class AsyncInvoker : public MessageHandler {
   template <class ReturnT, class FunctorT>
   void AsyncInvoke(const Location& posted_from,
                    Thread* thread,
-                   const FunctorT& functor,
+                   FunctorT&& functor,
                    uint32_t id = 0) {
     std::unique_ptr<AsyncClosure> closure(
-        new FireAndForgetAsyncClosure<FunctorT>(this, functor));
+        new FireAndForgetAsyncClosure<FunctorT>(
+            this, std::forward<FunctorT>(functor)));
     DoInvoke(posted_from, thread, std::move(closure), id);
   }
 
@@ -109,11 +110,12 @@ class AsyncInvoker : public MessageHandler {
   template <class ReturnT, class FunctorT>
   void AsyncInvokeDelayed(const Location& posted_from,
                           Thread* thread,
-                          const FunctorT& functor,
+                          FunctorT&& functor,
                           uint32_t delay_ms,
                           uint32_t id = 0) {
     std::unique_ptr<AsyncClosure> closure(
-        new FireAndForgetAsyncClosure<FunctorT>(this, functor));
+        new FireAndForgetAsyncClosure<FunctorT>(
+            this, std::forward<FunctorT>(functor)));
     DoInvokeDelayed(posted_from, thread, std::move(closure), delay_ms, id);
   }
 
@@ -188,12 +190,13 @@ class GuardedAsyncInvoker : public sigslot::has_slots<> {
   // immediately. Returns false if the thread has died.
   template <class ReturnT, class FunctorT>
   bool AsyncInvoke(const Location& posted_from,
-                   const FunctorT& functor,
+                   FunctorT&& functor,
                    uint32_t id = 0) {
     CritScope cs(&crit_);
     if (thread_ == nullptr)
       return false;
-    invoker_.AsyncInvoke<ReturnT, FunctorT>(posted_from, thread_, functor, id);
+    invoker_.AsyncInvoke<ReturnT, FunctorT>(
+        posted_from, thread_, std::forward<FunctorT>(functor), id);
     return true;
   }
 
@@ -201,14 +204,14 @@ class GuardedAsyncInvoker : public sigslot::has_slots<> {
   // completion. Returns immediately. Returns false if the thread has died.
   template <class ReturnT, class FunctorT>
   bool AsyncInvokeDelayed(const Location& posted_from,
-                          const FunctorT& functor,
+                          FunctorT&& functor,
                           uint32_t delay_ms,
                           uint32_t id = 0) {
     CritScope cs(&crit_);
     if (thread_ == nullptr)
       return false;
-    invoker_.AsyncInvokeDelayed<ReturnT, FunctorT>(posted_from, thread_,
+    invoker_.AsyncInvokeDelayed<ReturnT, FunctorT>(
-                                                   functor, delay_ms, id);
+        posted_from, thread_, std::forward<FunctorT>(functor), delay_ms, id);
     return true;
   }
 
@@ -217,7 +220,7 @@ class GuardedAsyncInvoker : public sigslot::has_slots<> {
   template <class ReturnT, class FunctorT, class HostT>
   bool AsyncInvoke(const Location& posted_from,
                    const Location& callback_posted_from,
-                   const FunctorT& functor,
+                   FunctorT&& functor,
                    void (HostT::*callback)(ReturnT),
                    HostT* callback_host,
                    uint32_t id = 0) {
@@ -225,8 +228,8 @@ class GuardedAsyncInvoker : public sigslot::has_slots<> {
     if (thread_ == nullptr)
       return false;
     invoker_.AsyncInvoke<ReturnT, FunctorT, HostT>(
-        posted_from, callback_posted_from, thread_, functor, callback,
+        posted_from, callback_posted_from, thread_,
-        callback_host, id);
+        std::forward<FunctorT>(functor), callback, callback_host, id);
     return true;
   }
 
@@ -235,7 +238,7 @@ class GuardedAsyncInvoker : public sigslot::has_slots<> {
   template <class ReturnT, class FunctorT, class HostT>
   bool AsyncInvoke(const Location& posted_from,
                    const Location& callback_posted_from,
-                   const FunctorT& functor,
+                   FunctorT&& functor,
                    void (HostT::*callback)(),
                    HostT* callback_host,
                    uint32_t id = 0) {
@@ -243,8 +246,8 @@ class GuardedAsyncInvoker : public sigslot::has_slots<> {
     if (thread_ == nullptr)
       return false;
     invoker_.AsyncInvoke<ReturnT, FunctorT, HostT>(
-        posted_from, callback_posted_from, thread_, functor, callback,
+        posted_from, callback_posted_from, thread_,
-        callback_host, id);
+        std::forward<FunctorT>(functor), callback, callback_host, id);
     return true;
   }
 

rtc_base/thread_unittest.cc

@@ -186,6 +186,16 @@ struct FunctorC {
     return 24;
   }
 };
+struct FunctorD {
+ public:
+  explicit FunctorD(AtomicBool* flag) : flag_(flag) {}
+  FunctorD(FunctorD&&) = default;
+  FunctorD& operator=(FunctorD&&) = default;
+  void operator()() { if (flag_) *flag_ = true; }
+ private:
+  AtomicBool* flag_;
+  RTC_DISALLOW_COPY_AND_ASSIGN(FunctorD);
+};
 
 // See: https://code.google.com/p/webrtc/issues/detail?id=2409
 TEST(ThreadTest, DISABLED_Main) {
@@ -441,6 +451,18 @@ TEST_F(AsyncInvokeTest, FireAndForget) {
   thread->Stop();
 }
 
+TEST_F(AsyncInvokeTest, NonCopyableFunctor) {
+  AsyncInvoker invoker;
+  // Create and start the thread.
+  auto thread = Thread::CreateWithSocketServer();
+  thread->Start();
+  // Try calling functor.
+  AtomicBool called;
+  invoker.AsyncInvoke<void>(RTC_FROM_HERE, thread.get(), FunctorD(&called));
+  EXPECT_TRUE_WAIT(called.get(), kWaitTimeout);
+  thread->Stop();
+}
+
 TEST_F(AsyncInvokeTest, KillInvokerDuringExecute) {
   // Use these events to get in a state where the functor is in the middle of
   // executing, and then to wait for it to finish, ensuring the "EXPECT_FALSE"
@@ -602,6 +624,14 @@ TEST_F(GuardedAsyncInvokeTest, FireAndForget) {
   EXPECT_TRUE_WAIT(called.get(), kWaitTimeout);
 }
 
+TEST_F(GuardedAsyncInvokeTest, NonCopyableFunctor) {
+  GuardedAsyncInvoker invoker;
+  // Try calling functor.
+  AtomicBool called;
+  EXPECT_TRUE(invoker.AsyncInvoke<void>(RTC_FROM_HERE, FunctorD(&called)));
+  EXPECT_TRUE_WAIT(called.get(), kWaitTimeout);
+}
+
 TEST_F(GuardedAsyncInvokeTest, Flush) {
   GuardedAsyncInvoker invoker;
   AtomicBool flag1;