Convert several backup container files to standard coroutines

fdbclient/AsyncFileS3BlobStore.actor.cpp → fdbclient/AsyncFileS3BlobStore.cpp

@@ -1,5 +1,5 @@
 /*
- * AsyncFileS3BlobStore.actor.cpp
+ * AsyncFileS3BlobStore.cpp
  *
  * This source file is part of the FoundationDB open source project
  *
@@ -22,7 +22,6 @@
 #include "fdbrpc/AsyncFileReadAhead.actor.h"
 #include "flow/UnitTest.h"
 #include "flow/IConnection.h"
-#include "flow/actorcompiler.h" // has to be last include
 
 Future<int64_t> AsyncFileS3BlobStoreRead::size() const {
 	if (!m_size.isValid())
@@ -34,34 +33,32 @@ Future<int> AsyncFileS3BlobStoreRead::read(void* data, int length, int64_t offse
 	return m_bstore->readObject(m_bucket, m_object, data, length, offset);
 }
 
-ACTOR Future<Void> sendStuff(int id, Reference<IRateControl> t, int bytes) {
+Future<Void> sendStuff(int id, Reference<IRateControl> t, int bytes) {
 	printf("Starting fake sender %d which will send send %d bytes.\n", id, bytes);
-	state double ts = timer();
-	state int total = 0;
+	double ts = timer();
+	int total = 0;
 	while (total < bytes) {
-		state int r = std::min<int>(deterministicRandom()->randomInt(0, 1000), bytes - total);
-		wait(t->getAllowance(r));
+		int r = std::min<int>(deterministicRandom()->randomInt(0, 1000), bytes - total);
+		co_await t->getAllowance(r);
 		total += r;
 	}
 	double dur = timer() - ts;
 	printf("Sender %d: Sent %d in %fs, %f/s\n", id, total, dur, total / dur);
-	return Void();
 }
 
 TEST_CASE("/backup/throttling") {
 	// Test will not work in simulation.
 	if (g_network->isSimulated())
-		return Void();
+		co_return;
 
-	state int limit = 100000;
-	state Reference<IRateControl> t(new SpeedLimit(limit, 1));
+	int limit = 100000;
+	Reference<IRateControl> t(new SpeedLimit(limit, 1));
 
-	state int id = 1;
+	int id = 1;
 	std::vector<Future<Void>> f;
-	state double ts = timer();
-	state int total = 0;
-	int s;
-	s = 500000;
+	double ts = timer();
+	int total = 0;
+	int s = 500000;
 	f.push_back(sendStuff(id++, t, s));
 	total += s;
 	f.push_back(sendStuff(id++, t, s));
@@ -75,11 +72,9 @@ TEST_CASE("/backup/throttling") {
 	f.push_back(sendStuff(id++, t, s));
 	total += s;
 
-	wait(waitForAll(f));
+	co_await waitForAll(f);
 	double dur = timer() - ts;
 	int speed = int(total / dur);
 	printf("Speed limit was %d, measured speed was %d\n", limit, speed);
 	ASSERT(abs(speed - limit) / limit < .01);
-
-	return Void();
 }

fdbclient/AsyncTaskThread.actor.cpp → fdbclient/AsyncTaskThread.cpp

@@ -1,5 +1,5 @@
 /*
- * AsyncTaskThread.actor.cpp
+ * AsyncTaskThread.cpp
  *
  * This source file is part of the FoundationDB open source project
  *
@@ -22,7 +22,6 @@
 
 #include "fdbclient/AsyncTaskThread.h"
 #include "flow/UnitTest.h"
-#include "flow/actorcompiler.h" // This must be the last #include.
 
 namespace {
 
@@ -32,28 +31,26 @@ class TerminateTask final : public IAsyncTask {
 	bool isTerminate() const override { return true; }
 };
 
-ACTOR Future<Void> asyncTaskThreadClient(AsyncTaskThread* asyncTaskThread,
-                                         std::atomic<int>* sum,
-                                         int count,
-                                         int clientId,
-                                         double meanSleep) {
-	state int i = 0;
-	state double randomSleep = 0.0;
-	for (; i < count; ++i) {
+Future<Void> asyncTaskThreadClient(AsyncTaskThread* asyncTaskThread,
+                                   std::atomic<int>* sum,
+                                   int count,
+                                   int clientId,
+                                   double meanSleep) {
+	double randomSleep = 0.0;
+	for (int i = 0; i < count; ++i) {
 		randomSleep = deterministicRandom()->random01() * 2 * meanSleep;
-		wait(delay(randomSleep));
-		wait(asyncTaskThread->execAsync([sum = sum] {
+		co_await delay(randomSleep);
+		co_await asyncTaskThread->execAsync([sum] {
 			sum->fetch_add(1);
 			return Void();
-		}));
+		});
 		TraceEvent("AsyncTaskThreadIncrementedSum")
 		    .detail("Index", i)
 		    .detail("Sum", sum->load())
 		    .detail("ClientId", clientId)
 		    .detail("RandomSleep", randomSleep)
 		    .detail("MeanSleep", meanSleep);
 	}
-	return Void();
 }
 
 } // namespace
@@ -90,31 +87,29 @@ void AsyncTaskThread::run(AsyncTaskThread* self) {
 }
 
 TEST_CASE("/asynctaskthread/add") {
-	state std::atomic<int> sum = 0;
-	state AsyncTaskThread asyncTaskThread;
-	state int numClients = 10;
-	state int incrementsPerClient = 100;
+	std::atomic<int> sum = 0;
+	AsyncTaskThread asyncTaskThread;
+	int numClients = 10;
+	int incrementsPerClient = 100;
 	std::vector<Future<Void>> clients;
 	clients.reserve(numClients);
 	for (int clientId = 0; clientId < numClients; ++clientId) {
 		clients.push_back(asyncTaskThreadClient(
 		    &asyncTaskThread, &sum, incrementsPerClient, clientId, deterministicRandom()->random01() * 0.01));
 	}
-	wait(waitForAll(clients));
+	co_await waitForAll(clients);
 	ASSERT_EQ(sum.load(), numClients * incrementsPerClient);
-	return Void();
 }
 
 TEST_CASE("/asynctaskthread/error") {
-	state AsyncTaskThread asyncTaskThread;
+	AsyncTaskThread asyncTaskThread;
 	try {
-		wait(asyncTaskThread.execAsync([] {
+		co_await asyncTaskThread.execAsync([] {
 			throw operation_failed();
 			return Void();
-		}));
+		});
 		ASSERT(false);
 	} catch (Error& e) {
 		ASSERT_EQ(e.code(), error_code_operation_failed);
 	}
-	return Void();
 }

fdbclient/BackupContainer.actor.cpp → fdbclient/BackupContainer.cpp

@@ -1,5 +1,5 @@
 /*
- * BackupContainer.actor.cpp
+ * BackupContainer.cpp
  *
  * This source file is part of the FoundationDB open source project
  *
@@ -51,15 +51,13 @@
 #include <algorithm>
 #include <cinttypes>
 #include <time.h>
-#include "flow/actorcompiler.h" // has to be last include
 
 namespace IBackupFile_impl {
 
-ACTOR Future<Void> appendStringRefWithLen(Reference<IBackupFile> file, Standalone<StringRef> s) {
-	state uint32_t lenBuf = bigEndian32((uint32_t)s.size());
-	wait(file->append(&lenBuf, sizeof(lenBuf)));
-	wait(file->append(s.begin(), s.size()));
-	return Void();
+Future<Void> appendStringRefWithLen(Reference<IBackupFile> file, Standalone<StringRef> s) {
+	uint32_t lenBuf = bigEndian32((uint32_t)s.size());
+	co_await file->append(&lenBuf, sizeof(lenBuf));
+	co_await file->append(s.begin(), s.size());
 }
 
 } // namespace IBackupFile_impl
@@ -387,12 +385,12 @@ Reference<IBackupContainer> IBackupContainer::openContainer(const std::string& u
 
 // Get a list of URLS to backup containers based on some a shorter URL.  This function knows about some set of supported
 // URL types which support this sort of backup discovery.
-ACTOR Future<std::vector<std::string>> listContainers_impl(std::string baseURL, Optional<std::string> proxy) {
+Future<std::vector<std::string>> listContainers_impl(std::string baseURL, Optional<std::string> proxy) {
 	try {
 		StringRef u(baseURL);
 		if (u.startsWith("file://"_sr)) {
-			std::vector<std::string> results = wait(BackupContainerLocalDirectory::listURLs(baseURL));
-			return results;
+			std::vector<std::string> results = co_await BackupContainerLocalDirectory::listURLs(baseURL);
+			co_return results;
 		} else if (u.startsWith("blobstore://"_sr)) {
 			std::string resource;
 
@@ -410,8 +408,8 @@ ACTOR Future<std::vector<std::string>> listContainers_impl(std::string baseURL,
 			// Create a dummy container to parse the backup-specific parameters from the URL and get a final bucket name
 			BackupContainerS3BlobStore dummy(bstore, "dummy", backupParams, {}, true);
 
-			std::vector<std::string> results = wait(BackupContainerS3BlobStore::listURLs(bstore, dummy.getBucket()));
-			return results;
+			std::vector<std::string> results = co_await BackupContainerS3BlobStore::listURLs(bstore, dummy.getBucket());
+			co_return results;
 		}
 		// TODO: Enable this when Azure backups are ready
 		/*
@@ -445,27 +443,28 @@ Future<std::vector<std::string>> IBackupContainer::listContainers(const std::str
 	return listContainers_impl(baseURL, proxy);
 }
 
-ACTOR Future<Version> timeKeeperVersionFromDatetime(std::string datetime, Database db) {
-	state KeyBackedMap<int64_t, Version> versionMap(timeKeeperPrefixRange.begin);
-	state Reference<ReadYourWritesTransaction> tr = makeReference<ReadYourWritesTransaction>(db);
+Future<Version> timeKeeperVersionFromDatetime(std::string datetime, Database db) {
+	KeyBackedMap<int64_t, Version> versionMap(timeKeeperPrefixRange.begin);
+	Reference<ReadYourWritesTransaction> tr = makeReference<ReadYourWritesTransaction>(db);
 
-	state int64_t time = BackupAgentBase::parseTime(datetime);
+	int64_t time = BackupAgentBase::parseTime(datetime);
 	if (time < 0) {
 		fprintf(
 		    stderr, "ERROR: Incorrect date/time or format.  Format is %s.\n", BackupAgentBase::timeFormat().c_str());
 		throw backup_error();
 	}
 
-	loop {
+	while (true) {
+		Error err;
 		try {
 			tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
 			tr->setOption(FDBTransactionOptions::LOCK_AWARE);
-			state KeyBackedRangeResult<std::pair<int64_t, Version>> rangeResult =
-			    wait(versionMap.getRange(tr, 0, time, 1, Snapshot::False, Reverse::True));
+			KeyBackedRangeResult<std::pair<int64_t, Version>> rangeResult =
+			    co_await versionMap.getRange(tr, 0, time, 1, Snapshot::False, Reverse::True);
 			if (rangeResult.results.size() != 1) {
 				// No key less than time was found in the database
 				// Look for a key >= time.
-				wait(store(rangeResult, versionMap.getRange(tr, time, std::numeric_limits<int64_t>::max(), 1)));
+				co_await store(rangeResult, versionMap.getRange(tr, time, std::numeric_limits<int64_t>::max(), 1));
 
 				if (rangeResult.results.size() != 1) {
 					fprintf(stderr, "ERROR: Unable to calculate a version for given date/time.\n");
@@ -475,42 +474,43 @@ ACTOR Future<Version> timeKeeperVersionFromDatetime(std::string datetime, Databa
 
 			// Adjust version found by the delta between time and the time found and min with 0.
 			auto& result = rangeResult.results[0];
-			return std::max<Version>(0, result.second + (time - result.first) * CLIENT_KNOBS->CORE_VERSIONSPERSECOND);
-
+			co_return std::max<Version>(0,
+			                            result.second + (time - result.first) * CLIENT_KNOBS->CORE_VERSIONSPERSECOND);
 		} catch (Error& e) {
-			wait(tr->onError(e));
+			err = e;
 		}
+		co_await tr->onError(err);
 	}
 }
 
-ACTOR Future<Optional<int64_t>> timeKeeperEpochsFromVersion(Version v, Reference<ReadYourWritesTransaction> tr) {
-	state KeyBackedMap<int64_t, Version> versionMap(timeKeeperPrefixRange.begin);
+Future<Optional<int64_t>> timeKeeperEpochsFromVersion(Version v, Reference<ReadYourWritesTransaction> tr) {
+	KeyBackedMap<int64_t, Version> versionMap(timeKeeperPrefixRange.begin);
 
 	// Binary search to find the closest date with a version <= v
-	state int64_t min = 0;
-	state int64_t max = (int64_t)now();
-	state int64_t mid;
-	state std::pair<int64_t, Version> found;
+	int64_t min = 0;
+	int64_t max = (int64_t)now();
+	int64_t mid{ 0 };
+	std::pair<int64_t, Version> found;
 
 	tr->setOption(FDBTransactionOptions::ACCESS_SYSTEM_KEYS);
 	tr->setOption(FDBTransactionOptions::LOCK_AWARE);
 
-	loop {
+	while (true) {
 		mid = (min + max + 1) / 2; // ceiling
 
 		// Find the highest time < mid
-		state KeyBackedRangeResult<std::pair<int64_t, Version>> rangeResult =
-		    wait(versionMap.getRange(tr, min, mid, 1, Snapshot::False, Reverse::True));
+		KeyBackedRangeResult<std::pair<int64_t, Version>> rangeResult =
+		    co_await versionMap.getRange(tr, min, mid, 1, Snapshot::False, Reverse::True);
 
 		if (rangeResult.results.size() != 1) {
 			if (mid == min) {
 				// There aren't any records having a version < v, so just look for any record having a time < now
 				// and base a result on it
-				wait(store(rangeResult, versionMap.getRange(tr, 0, (int64_t)now(), 1)));
+				co_await store(rangeResult, versionMap.getRange(tr, 0, (int64_t)now(), 1));
 
 				if (rangeResult.results.size() != 1) {
 					// There aren't any timekeeper records to base a result on so return nothing
-					return Optional<int64_t>();
+					co_return Optional<int64_t>();
 				}
 
 				found = rangeResult.results[0];
@@ -533,5 +533,5 @@ ACTOR Future<Optional<int64_t>> timeKeeperEpochsFromVersion(Version v, Reference
 		}
 	}
 
-	return found.first + (v - found.second) / CLIENT_KNOBS->CORE_VERSIONSPERSECOND;
+	co_return found.first + (v - found.second) / CLIENT_KNOBS->CORE_VERSIONSPERSECOND;
 }