Merge pull request #2915 from tclinken/move-optimizations

Avoid unnecessary copies in PromiseStream

fdbrpc/FlowTests.actor.cpp

@@ -1360,3 +1360,94 @@ TEST_CASE("/flow/DeterministicRandom/SignedOverflow") {
 	       std::numeric_limits<int64_t>::max() - 1);
 	return Void();
 }
+
+struct Tracker {
+	int copied;
+	bool moved;
+	Tracker(int copied = 0) : moved(false), copied(copied) {}
+	Tracker(Tracker&& other) : Tracker(other.copied) {
+		ASSERT(!other.moved);
+		other.moved = true;
+	}
+	Tracker& operator=(Tracker&& other) {
+		ASSERT(!other.moved);
+		other.moved = true;
+		this->moved = false;
+		this->copied = other.copied;
+		return *this;
+	}
+	Tracker(const Tracker& other) : Tracker(other.copied + 1) { ASSERT(!other.moved); }
+	Tracker& operator=(const Tracker& other) {
+		ASSERT(!other.moved);
+		this->moved = false;
+		this->copied = other.copied + 1;
+		return *this;
+	}
+
+	ACTOR static Future<Void> listen(FutureStream<Tracker> stream) {
+		Tracker t = waitNext(stream);
+		ASSERT(!t.moved);
+		ASSERT(t.copied == 0);
+		return Void();
+	}
+};
+
+TEST_CASE("/flow/flow/PromiseStream/move") {
+	state PromiseStream<Tracker> stream;
+	{
+		// This tests the case when a callback is added before
+		// a movable value is sent
+		state Future<Void> listener = Tracker::listen(stream.getFuture());
+		stream.send(Tracker{});
+		wait(listener);
+	}
+
+	{
+		// This tests the case when a callback is added before
+		// a unmovable value is sent
+		listener = Tracker::listen(stream.getFuture());
+		Tracker namedTracker;
+		stream.send(namedTracker);
+		wait(listener);
+	}
+	{
+		// This tests the case when no callback is added until
+		// after a movable value is sent
+		stream.send(Tracker{});
+		stream.send(Tracker{});
+		{
+			Tracker t = waitNext(stream.getFuture());
+			ASSERT(!t.moved);
+			ASSERT(t.copied == 0);
+		}
+		choose {
+			when(Tracker t = waitNext(stream.getFuture())) {
+				ASSERT(!t.moved);
+				ASSERT(t.copied == 0);
+			}
+		}
+	}
+	{
+		// This tests the case when no callback is added until
+		// after an unmovable value is sent
+		Tracker namedTracker1;
+		Tracker namedTracker2;
+		stream.send(namedTracker1);
+		stream.send(namedTracker2);
+		{
+			Tracker t = waitNext(stream.getFuture());
+			ASSERT(!t.moved);
+			// must copy onto queue
+			ASSERT(t.copied == 1);
+		}
+		choose {
+			when(Tracker t = waitNext(stream.getFuture())) {
+				ASSERT(!t.moved);
+				// must copy onto queue
+				ASSERT(t.copied == 1);
+			}
+		}
+	}
+
+	return Void();
+}

fdbrpc/fdbrpc.h

@@ -247,6 +247,15 @@ class RequestStream {
 		else
 			queue->send(value);
 	}
+
+	void send(T&& value) const {
+		if (queue->isRemoteEndpoint()) {
+			FlowTransport::transport().sendUnreliable(SerializeSource<T>(std::move(value)), getEndpoint(), true);
+		}
+		else
+			queue->send(std::move(value));
+	}
+
 	/*void sendError(const Error& error) const {
 	ASSERT( !queue->isRemoteEndpoint() );
 	queue->sendError(error);

fdbserver/MasterProxyServer.actor.cpp

@@ -508,7 +508,7 @@ ACTOR Future<Void> commitBatcher(ProxyCommitData *commitData, PromiseStream<std:
 					}
 
 					if((batchBytes + bytes > CLIENT_KNOBS->TRANSACTION_SIZE_LIMIT || req.firstInBatch()) && batch.size()) {
-						out.send({ batch, batchBytes });
+						out.send({ std::move(batch), batchBytes });
 						lastBatch = now();
 						timeout = delayJittered(commitData->commitBatchInterval, TaskPriority::ProxyCommitBatcher);
 						batch = std::vector<CommitTransactionRequest>();

flow/actorcompiler/ActorCompiler.cs

@@ -813,7 +813,7 @@ void CompileStatement(ChooseStatement stmt, Context cx)
                     returnType = "void",
                     formalParameters = new string[] { 
                         ch.CallbackTypeInStateClass + "*",
-                        ch.Stmt.wait.result.type + " value"
+                        ch.Stmt.wait.result.type + " const& value"
                     },
                     endIsUnreachable = true
                 };

flow/flow.h

@@ -586,7 +586,7 @@ struct NotifiedQueue : private SingleCallback<T>, FastAllocated<NotifiedQueue<T>
 			if (error.isValid()) throw error;
 			throw internal_error();
 		}
-		auto copy = queue.front();
+		auto copy = std::move(queue.front());
 		queue.pop();
 		return copy;
 	}
@@ -908,6 +908,9 @@ class PromiseStream {
 	void send(const T& value) const {
 		queue->send(value);
 	}
+	void send(T&& value) const {
+		queue->send(std::move(value));
+	}
 	void sendError(const Error& error) const {
 		queue->sendError(error);
 	}