stub changes for IO queue and attempt at new benchmark

CMakeLists.txt

@@ -180,7 +180,7 @@ endif()
 add_subdirectory(fdbbackup)
 add_subdirectory(contrib)
 add_subdirectory(tests)
-add_subdirectory(flowbench EXCLUDE_FROM_ALL)
+add_subdirectory(flowbench)
 if(WITH_PYTHON AND WITH_C_BINDING)
   add_subdirectory(bindings)
 endif()

flow/AsioReactor.h

@@ -39,6 +39,7 @@ public:
 
 	void sleep(double timeout);
 	void react();
+	void react_block();
 
 	void wake();
 

flow/Net2.actor.cpp

@@ -1370,6 +1370,11 @@ ACTOR Future<Void> Net2::logTimeOffset() {
 	}
 }
 
+THREAD_FUNC_RETURN reactorThreadTest(void* reactor) {
+	loop { ((ASIOReactor*)reactor)->react_block(); }
+	THREAD_RETURN;
+}
+
 void Net2::initMetrics() {
 	bytesReceived.init(LiteralStringRef("Net2.BytesReceived"));
 	countWriteProbes.init(LiteralStringRef("Net2.CountWriteProbes"));
@@ -1432,6 +1437,7 @@ void Net2::run() {
 
 	started.store(true);
 	double nnow = timer_monotonic();
+	// startThread(reactorThreadTest, (void*)&reactor, 0, 0);
 
 	while (!stopped) {
 		FDB_TRACE_PROBE(run_loop_begin);
@@ -1479,6 +1485,8 @@ void Net2::run() {
 		tscBegin = timestampCounter();
 		taskBegin = timer_monotonic();
 		trackAtPriority(TaskPriority::ASIOReactor, taskBegin);
+		// Reactor pushes IO Futures as ready and moves associated tasks
+		// into ready queue
 		reactor.react();
 
 		updateNow();
@@ -1499,9 +1507,11 @@ void Net2::run() {
 			ready.push(timers.top());
 			timers.pop();
 		}
+		// is this double counting?
 		countTimers += numTimers;
 		FDB_TRACE_PROBE(run_loop_ready_timers, numTimers);
 
+		// moves items from threadReady queue to ready queue
 		processThreadReady();
 
 		tscBegin = timestampCounter();
@@ -1513,12 +1523,16 @@ void Net2::run() {
 
 		FDB_TRACE_PROBE(run_loop_tasks_start, queueSize);
 		while (!ready.empty()) {
+			// TraceEvent("DebugRunLoopReadyStart")
+			// 	.detail("ReadySize", ready.size());
+			// aiming to empty the ready queue
 			++countTasks;
 			currentTaskID = ready.top().taskID;
 			priorityMetric = static_cast<int64_t>(currentTaskID);
 			Task* task = ready.top().task;
 			ready.pop();
 
+			// perform highest prio task in ready queue
 			try {
 				(*task)();
 			} catch (Error& e) {
@@ -1532,8 +1546,15 @@ void Net2::run() {
 				minTaskID = currentTaskID;
 			}
 
+			// attempt to empty out the IO backlog for every X tasks while in this queue
+			// if (ready.size() % 5 == 0) {
+			// 	reactor.react();
+			// }
+
 			double tscNow = timestampCounter();
 			double newTaskBegin = timer_monotonic();
+			// break from emptying ready queue if tscNow is outside the bounds of [tscBegin, tscEnd]
+			// or if the next item in the prio queue is greater than given prio below
 			if (check_yield(TaskPriority::Max, tscNow)) {
 				checkForSlowTask(tscBegin, tscNow, newTaskBegin - taskBegin, currentTaskID);
 				taskBegin = newTaskBegin;
@@ -1545,6 +1566,7 @@ void Net2::run() {
 			taskBegin = newTaskBegin;
 			tscBegin = tscNow;
 		}
+		// TraceEvent("DebugRunLoopReadyEnd");
 
 		trackAtPriority(TaskPriority::RunLoop, taskBegin);
 
@@ -1601,7 +1623,7 @@ void Net2::run() {
 		    nondeterministicRandom()->random01() < (nnow - now) * FLOW_KNOBS->SLOW_LOOP_SAMPLING_RATE)
 			TraceEvent("SomewhatSlowRunLoopBottom")
 			    .detail("Elapsed", nnow - now); // This includes the time spent running tasks
-	}
+	} // while (!stopped)
 
 	for (auto& fn : stopCallbacks) {
 		fn();
@@ -1610,7 +1632,7 @@ void Net2::run() {
 #ifdef WIN32
 	timeEndPeriod(1);
 #endif
-}
+} // Net2::run
 
 // Updates the PriorityStats found in NetworkMetrics
 void Net2::updateStarvationTracker(struct NetworkMetrics::PriorityStats& binStats,
@@ -1990,6 +2012,12 @@ void ASIOReactor::react() {
 		++network->countASIOEvents; // Make this a task?
 }
 
+void ASIOReactor::react_block() {
+	ios.run_one();
+	ios.restart();
+	++network->countASIOEvents;
+}
+
 void ASIOReactor::wake() {
 	ios.post(nullCompletionHandler);
 }

flowbench/BenchIONet2.actor.cpp

@@ -0,0 +1,81 @@
+/*
+ * BenchIONet2.actor.cpp
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2013-2020 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "benchmark/benchmark.h"
+
+#include "flow/Platform.h"
+#include "flow/flow.h"
+#include "flow/DeterministicRandom.h"
+#include "flow/network.h"
+#include "flow/ThreadHelper.actor.h"
+#include "fdbrpc/IAsyncFile.h"
+
+#include "flow/actorcompiler.h" // This must be the last #include.
+
+ACTOR static Future<Void> increment(TaskPriority priority, uint32_t* sum) {
+	wait(delay(0, priority));
+	threadSleep(0.000001);
+	++(*sum);
+	return Void();
+}
+
+static inline TaskPriority getRandomTaskPriority(DeterministicRandom& rand) {
+	return static_cast<TaskPriority>(rand.randomInt(0, 100));
+}
+
+ACTOR static Future<Void> benchIONet2Actor(benchmark::State* benchState) {
+	state size_t actorCount = benchState->range(0);
+	state uint32_t sum;
+	state int seed = platform::getRandomSeed();
+	state std::unique_ptr<char[]> data(new char[4096]);
+	memset(data.get(), 0, 4096);
+	while (benchState->KeepRunning()) {
+		sum = 0;
+		state std::vector<Future<Void>> futures;
+		futures.reserve(actorCount);
+		DeterministicRandom rand(seed);
+		// for (int i = 0; i < 1; ++i) {
+		// 	futures.push_back(increment(getRandomTaskPriority(rand), &sum));
+		// }
+		printf("1\n");
+		state Reference<IAsyncFile> f =
+		    wait(IAsyncFileSystem::filesystem()->open("test-benchmark-file", IAsyncFile::OPEN_READWRITE, 0600));
+
+		printf("2\n");
+
+		wait(f->sync());
+
+		printf("3\n");
+		// wait(f->truncate(100e6));
+		// wait(f->write(data.get(), 4096, 0));
+		// wait(f->sync());
+		// f.clear();
+		// wait(waitForAll(futures));
+		benchmark::DoNotOptimize(sum);
+	}
+	benchState->SetItemsProcessed(actorCount * static_cast<long>(benchState->iterations()));
+	return Void();
+}
+
+static void bench_ionet2(benchmark::State& benchState) {
+	onMainThread([&benchState] { return benchIONet2Actor(&benchState); }).blockUntilReady();
+}
+
+BENCHMARK(bench_ionet2)->Range(1, 1 << 16)->ReportAggregatesOnly(true);

flowbench/CMakeLists.txt

@@ -3,6 +3,7 @@ set(FLOWBENCH_SRCS
   BenchCallback.actor.cpp
   BenchHash.cpp
   BenchIterate.cpp
+  BenchIONet2.actor.cpp
   BenchMem.cpp
   BenchMetadataCheck.cpp
   BenchNet2.actor.cpp