common: forward_handler() for handlers with move-only args

executors will always call the handler's lvalue overload of operator(),
so forward_handler() is required to wrap calls to bind_handler() with
move-only argument types. Completion uses this to forward all of its
CompletionHandlers

Signed-off-by: Casey Bodley <cbodley@redhat.com>

src/common/async/completion.h

@@ -18,6 +18,7 @@
 #include <memory>
 
 #include "bind_handler.h"
+#include "forward_handler.h"
 
 namespace ceph::async {
 
@@ -180,9 +181,13 @@ class CompletionImpl final : public Completion<void(Args...), T> {
     RebindTraits2::deallocate(alloc2, static_cast<CompletionImpl*>(p), 1);
   }
 
+  static auto bind_and_forward(Handler&& h, std::tuple<Args...>&& args) {
+    return forward_handler(CompletionHandler{std::move(h), std::move(args)});
+  }
+
   void destroy_defer(std::tuple<Args...>&& args) override {
     auto w = std::move(work);
-    auto f = CompletionHandler{std::move(handler), std::move(args)};
+    auto f = bind_and_forward(std::move(handler), std::move(args));
     RebindAlloc2 alloc2 = boost::asio::get_associated_allocator(handler);
     RebindTraits2::destroy(alloc2, this);
     RebindTraits2::deallocate(alloc2, this, 1);
@@ -190,7 +195,7 @@ class CompletionImpl final : public Completion<void(Args...), T> {
   }
   void destroy_dispatch(std::tuple<Args...>&& args) override {
     auto w = std::move(work);
-    auto f = CompletionHandler{std::move(handler), std::move(args)};
+    auto f = bind_and_forward(std::move(handler), std::move(args));
     RebindAlloc2 alloc2 = boost::asio::get_associated_allocator(handler);
     RebindTraits2::destroy(alloc2, this);
     RebindTraits2::deallocate(alloc2, this, 1);
@@ -198,7 +203,7 @@ class CompletionImpl final : public Completion<void(Args...), T> {
   }
   void destroy_post(std::tuple<Args...>&& args) override {
     auto w = std::move(work);
-    auto f = CompletionHandler{std::move(handler), std::move(args)};
+    auto f = bind_and_forward(std::move(handler), std::move(args));
     RebindAlloc2 alloc2 = boost::asio::get_associated_allocator(handler);
     RebindTraits2::destroy(alloc2, this);
     RebindTraits2::deallocate(alloc2, this, 1);

src/common/async/forward_handler.h

@@ -0,0 +1,103 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2018 Red Hat
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#ifndef CEPH_ASYNC_FORWARD_HANDLER_H
+#define CEPH_ASYNC_FORWARD_HANDLER_H
+
+#include <boost/asio.hpp>
+
+namespace ceph::async {
+
+/**
+ * A forwarding completion handler for use with boost::asio.
+ *
+ * A completion handler wrapper that invokes the handler's operator() as an
+ * rvalue, regardless of whether the wrapper is invoked as an lvalue or rvalue.
+ * This operation is potentially destructive to the wrapped handler, so is only
+ * suitable for single-use handlers.
+ *
+ * This is useful when combined with bind_handler() and move-only arguments,
+ * because executors will always call the lvalue overload of operator().
+ *
+ * The original Handler's associated allocator and executor are maintained.
+ *
+ * @see forward_handler
+ */
+template <typename Handler>
+struct ForwardingHandler {
+  Handler handler;
+
+  ForwardingHandler(Handler&& handler)
+    : handler(std::move(handler))
+  {}
+
+  template <typename ...Args>
+  void operator()(Args&& ...args) {
+    std::move(handler)(std::forward<Args>(args)...);
+  }
+
+  using allocator_type = boost::asio::associated_allocator_t<Handler>;
+  allocator_type get_allocator() const noexcept {
+    return boost::asio::get_associated_allocator(handler);
+  }
+};
+
+} // namespace ceph::async
+
+namespace boost::asio {
+
+// specialize boost::asio::associated_executor<> for ForwardingHandler
+template <typename Handler, typename Executor>
+struct associated_executor<ceph::async::ForwardingHandler<Handler>, Executor> {
+  using type = boost::asio::associated_executor_t<Handler, Executor>;
+
+  static type get(const ceph::async::ForwardingHandler<Handler>& handler,
+                  const Executor& ex = Executor()) noexcept {
+    return boost::asio::get_associated_executor(handler.handler, ex);
+  }
+};
+
+} // namespace boost::asio
+
+namespace ceph::async {
+
+/**
+ * Returns a single-use completion handler that always forwards on operator().
+ *
+ * Wraps a completion handler such that it is always invoked as an rvalue. This
+ * is necessary when combining executors and bind_handler() with move-only
+ * argument types.
+ *
+ * Example use:
+ *
+ *   auto callback = [] (std::unique_ptr<int>&& p) {};
+ *   auto bound_handler = bind_handler(callback, std::make_unique<int>(5));
+ *   auro handler = forward_handler(std::move(bound_handler));
+ *
+ *   // execute the forwarding handler on an io_context:
+ *   boost::asio::io_context context;
+ *   boost::asio::post(context, std::move(handler));
+ *   context.run();
+ *
+ * @see ForwardingHandler
+ */
+template <typename Handler>
+auto forward_handler(Handler&& h)
+{
+  return ForwardingHandler{std::forward<Handler>(h)};
+}
+
+} // namespace ceph::async
+
+#endif // CEPH_ASYNC_FORWARD_HANDLER_H

src/test/common/test_async_completion.cc

@@ -43,7 +43,7 @@ TEST(AsyncCompletion, BindHandler)
   auto b2 = bind_handler(std::move(h2), move_only{});
   std::move(b2)();
 
-  // references bound with std::ref() and passed to all operator() overloads
+  // references bound with std::ref() can be passed to all operator() overloads
   auto h3 = [] (int& c) { c++; };
   int count = 0;
   auto b3 = bind_handler(std::move(h3), std::ref(count));
@@ -57,12 +57,22 @@ TEST(AsyncCompletion, BindHandler)
   EXPECT_EQ(3, count);
 }
 
+TEST(AsyncCompletion, ForwardHandler)
+{
+  // move-only types can be forwarded with 'operator() &'
+  auto h = [] (move_only&& m) {};
+  auto b = bind_handler(std::move(h), move_only{});
+  auto f = forward_handler(std::move(b));
+  f();
+}
+
 TEST(AsyncCompletion, MoveOnly)
 {
   boost::asio::io_context context;
   auto ex1 = context.get_executor();
 
-  std::optional<error_code> ec1, ec2;
+  std::optional<error_code> ec1, ec2, ec3;
+  std::optional<move_only> arg3;
   {
     // move-only user data
     using Completion = Completion<void(error_code), move_only>;
@@ -77,6 +87,16 @@ TEST(AsyncCompletion, MoveOnly)
     Completion::post(std::move(c), boost::asio::error::operation_aborted);
     EXPECT_FALSE(ec2);
   }
+  {
+    // move-only arg in signature
+    using Completion = Completion<void(error_code, move_only)>;
+    auto c = Completion::create(ex1, [&] (error_code ec, move_only m) {
+        ec3 = ec;
+        arg3 = std::move(m);
+      });
+    Completion::post(std::move(c), boost::asio::error::operation_aborted, move_only{});
+    EXPECT_FALSE(ec3);
+  }
 
   context.poll();
   EXPECT_TRUE(context.stopped());
@@ -85,6 +105,9 @@ TEST(AsyncCompletion, MoveOnly)
   EXPECT_EQ(boost::asio::error::operation_aborted, *ec1);
   ASSERT_TRUE(ec2);
   EXPECT_EQ(boost::asio::error::operation_aborted, *ec2);
+  ASSERT_TRUE(ec3);
+  EXPECT_EQ(boost::asio::error::operation_aborted, *ec3);
+  ASSERT_TRUE(arg3);
 }
 
 TEST(AsyncCompletion, VoidCompletion)