fix: preserve lvalues in execution queue

src/bthread/execution_queue_inl.h

@@ -302,7 +302,7 @@ friend class TaskIterator<T>;
 
     int execute(typename butil::add_const_reference<T>::type task,
                 const TaskOptions* options, TaskHandle* handle) {
-        return execute(std::forward<T>(const_cast<T&>(task)), options, handle);
+        return execute_impl(task, options, handle);
     }
 
 
@@ -312,6 +312,13 @@ friend class TaskIterator<T>;
 
     int execute(T&& task,
                 const TaskOptions* options, TaskHandle* handle) {
+        return execute_impl(std::move(task), options, handle);
+    }
+
+private:
+    template <typename U>
+    int execute_impl(U&& task,
+                     const TaskOptions* options, TaskHandle* handle) {
         if (stopped()) {
             return EINVAL;
         }
@@ -324,7 +331,7 @@ friend class TaskIterator<T>;
             return_task_node(node);
             return ENOMEM;
         }
-        new (mem) T(std::forward<T>(task));
+        new (mem) T(std::forward<U>(task));
         node->stop_task = false;
         TaskOptions opt;
         if (options) {

test/bthread_execution_queue_unittest.cpp

@@ -150,6 +150,47 @@ TEST_F(ExecutionQueueTest, rvalue) {
     }
 }
 
+class CopyMoveTask {
+public:
+    explicit CopyMoveTask(int value) : _value(value), _moved(false) {}
+
+    CopyMoveTask(const CopyMoveTask& rhs)
+        : _value(rhs._value), _moved(false) {}
+
+    CopyMoveTask(CopyMoveTask&& rhs) noexcept
+        : _value(rhs._value), _moved(false) {
+        rhs._moved = true;
+    }
+
+    int value() const { return _value; }
+    bool moved() const { return _moved; }
+
+private:
+    int _value;
+    bool _moved;
+};
+
+int add(void* meta, bthread::TaskIterator<CopyMoveTask>& iter) {
+    int* result = static_cast<int*>(meta);
+    for (; iter; ++iter) {
+        *result += iter->value();
+    }
+    return 0;
+}
+
+TEST_F(ExecutionQueueTest, lvalue_is_copied) {
+    bthread::ExecutionQueueId<CopyMoveTask> queue_id;
+    int result = 0;
+    ASSERT_EQ(0, bthread::execution_queue_start(&queue_id, nullptr,
+                                                add, &result));
+    CopyMoveTask task(42);
+    ASSERT_EQ(0, bthread::execution_queue_execute(queue_id, task));
+    EXPECT_FALSE(task.moved());
+    ASSERT_EQ(0, bthread::execution_queue_stop(queue_id));
+    ASSERT_EQ(0, bthread::execution_queue_join(queue_id));
+    EXPECT_EQ(42, result);
+}
+
 struct PushArg {
     bthread::ExecutionQueueId<LongIntTask> id {0};
     butil::atomic<int64_t> total_num {0};