From ebb570422bc8fcc14a800e61320e9aab946fff65 Mon Sep 17 00:00:00 2001
From: Andrew Noyes <andrew.noyes@snowflake.com>
Date: Wed, 22 Dec 2021 15:01:54 -0800
Subject: [PATCH 1/4] Format tmp_cluster.py with black
---
tests/TestRunner/tmp_cluster.py | 70 ++++++++++++++++++++++-----------
1 file changed, 46 insertions(+), 24 deletions(-)
diff --git a/tests/TestRunner/tmp_cluster.py b/tests/TestRunner/tmp_cluster.py
index 5c84f6086f..5181bce092 100755
--- a/tests/TestRunner/tmp_cluster.py
+++ b/tests/TestRunner/tmp_cluster.py
@@ -10,21 +10,25 @@ from argparse import ArgumentParser, RawDescriptionHelpFormatter
from random import choice
from pathlib import Path
+
class TempCluster:
def __init__(self, build_dir: str, process_number: int = 1, port: str = None):
self.build_dir = Path(build_dir).resolve()
assert self.build_dir.exists(), "{} does not exist".format(build_dir)
assert self.build_dir.is_dir(), "{} is not a directory".format(build_dir)
tmp_dir = self.build_dir.joinpath(
- 'tmp',
- ''.join(choice(LocalCluster.valid_letters_for_secret) for i in range(16)))
+ "tmp",
+ "".join(choice(LocalCluster.valid_letters_for_secret) for i in range(16)),
+ )
tmp_dir.mkdir(parents=True)
- self.cluster = LocalCluster(tmp_dir,
- self.build_dir.joinpath('bin', 'fdbserver'),
- self.build_dir.joinpath('bin', 'fdbmonitor'),
- self.build_dir.joinpath('bin', 'fdbcli'),
- process_number,
- port = port)
+ self.cluster = LocalCluster(
+ tmp_dir,
+ self.build_dir.joinpath("bin", "fdbserver"),
+ self.build_dir.joinpath("bin", "fdbmonitor"),
+ self.build_dir.joinpath("bin", "fdbcli"),
+ process_number,
+ port=port,
+ )
self.log = self.cluster.log
self.etc = self.cluster.etc
self.data = self.cluster.data
@@ -40,13 +44,14 @@ class TempCluster:
shutil.rmtree(self.tmp_dir)
def close(self):
- self.cluster.__exit__(None,None,None)
+ self.cluster.__exit__(None, None, None)
shutil.rmtree(self.tmp_dir)
-if __name__ == '__main__':
- parser = ArgumentParser(formatter_class=RawDescriptionHelpFormatter,
- description="""
+if __name__ == "__main__":
+ parser = ArgumentParser(
+ formatter_class=RawDescriptionHelpFormatter,
+ description="""
This script automatically configures a temporary local cluster on the machine
and then calls a command while this cluster is running. As soon as the command
returns, the configured cluster is killed and all generated data is deleted.
@@ -61,30 +66,47 @@ if __name__ == '__main__':
- All occurrences of @ETC_DIR@ will be replaced with the path to the configuration directory.
The environment variable FDB_CLUSTER_FILE is set to the generated cluster for the command if it is not set already.
- """)
- parser.add_argument('--build-dir', '-b', metavar='BUILD_DIRECTORY', help='FDB build directory', required=True)
- parser.add_argument('cmd', metavar="COMMAND", nargs="+", help="The command to run")
- parser.add_argument('--process-number', '-p', help="Number of fdb processes running", type=int, default=1)
+ """,
+ )
+ parser.add_argument(
+ "--build-dir",
+ "-b",
+ metavar="BUILD_DIRECTORY",
+ help="FDB build directory",
+ required=True,
+ )
+ parser.add_argument("cmd", metavar="COMMAND", nargs="+", help="The command to run")
+ parser.add_argument(
+ "--process-number",
+ "-p",
+ help="Number of fdb processes running",
+ type=int,
+ default=1,
+ )
args = parser.parse_args()
errcode = 1
with TempCluster(args.build_dir, args.process_number) as cluster:
print("log-dir: {}".format(cluster.log))
print("etc-dir: {}".format(cluster.etc))
print("data-dir: {}".format(cluster.data))
- print("cluster-file: {}".format(cluster.etc.joinpath('fdb.cluster')))
+ print("cluster-file: {}".format(cluster.etc.joinpath("fdb.cluster")))
cmd_args = []
for cmd in args.cmd:
- if cmd == '@CLUSTER_FILE@':
- cmd_args.append(str(cluster.etc.joinpath('fdb.cluster')))
- elif cmd == '@DATA_DIR@':
+ if cmd == "@CLUSTER_FILE@":
+ cmd_args.append(str(cluster.etc.joinpath("fdb.cluster")))
+ elif cmd == "@DATA_DIR@":
cmd_args.append(str(cluster.data))
- elif cmd == '@LOG_DIR@':
+ elif cmd == "@LOG_DIR@":
cmd_args.append(str(cluster.log))
- elif cmd == '@ETC_DIR@':
+ elif cmd == "@ETC_DIR@":
cmd_args.append(str(cluster.etc))
else:
cmd_args.append(cmd)
env = dict(**os.environ)
- env['FDB_CLUSTER_FILE'] = env.get('FDB_CLUSTER_FILE', cluster.etc.joinpath('fdb.cluster'))
- errcode = subprocess.run(cmd_args, stdout=sys.stdout, stderr=sys.stderr, env=env).returncode
+ env["FDB_CLUSTER_FILE"] = env.get(
+ "FDB_CLUSTER_FILE", cluster.etc.joinpath("fdb.cluster")
+ )
+ errcode = subprocess.run(
+ cmd_args, stdout=sys.stdout, stderr=sys.stderr, env=env
+ ).returncode
sys.exit(errcode)
From 6fdbd9ae30d47053ea283f0adc02412dcc4ebf47 Mon Sep 17 00:00:00 2001
From: Andrew Noyes <andrew.noyes@snowflake.com>
Date: Wed, 22 Dec 2021 12:26:47 -0800
Subject: [PATCH 2/4] Fail test if there's a sev40 in the logs, and output logs
if test fails
---
tests/TestRunner/tmp_cluster.py | 19 ++++++++++++++++++-
1 file changed, 18 insertions(+), 1 deletion(-)
diff --git a/tests/TestRunner/tmp_cluster.py b/tests/TestRunner/tmp_cluster.py
index 5181bce092..3ff23fe8c3 100755
--- a/tests/TestRunner/tmp_cluster.py
+++ b/tests/TestRunner/tmp_cluster.py
@@ -1,10 +1,10 @@
#!/usr/bin/env python3
+import glob
import os
import shutil
import subprocess
import sys
-import socket
from local_cluster import LocalCluster
from argparse import ArgumentParser, RawDescriptionHelpFormatter
from random import choice
@@ -109,4 +109,21 @@ if __name__ == "__main__":
errcode = subprocess.run(
cmd_args, stdout=sys.stdout, stderr=sys.stderr, env=env
).returncode
+ sev40s = (
+ subprocess.run(
+ ["grep", 'Severity="40"', os.path.join(cluster.log)],
+ stdout=sys.stdout,
+ stderr=sys.stderr,
+ ).returncode
+ != 0
+ )
+ if sev40s:
+ print(">>>>>>>>>>>>>>>>>>>> Found severity 40 events - the test fails")
+ errcode = 1
+ if errcode:
+ for log_file in glob.glob(os.path.join(cluster.log, "*")):
+ print(">>>>>>>>>>>>>>>>>>>> Contents of {}:".format(log_file))
+ with open(log_file, "r") as f:
+ print(f.read())
+
sys.exit(errcode)
From 28971c5181cdd55ac648508e098ccaed60344e25 Mon Sep 17 00:00:00 2001
From: Andrew Noyes <andrew.noyes@snowflake.com>
Date: Wed, 22 Dec 2021 14:59:06 -0800
Subject: [PATCH 3/4] Fix memory leak. Closes #4482
---
fdbrpc/fdbrpc.h | 2 ++
fdbrpc/networksender.actor.h | 4 ++++
flow/error_definitions.h | 1 +
3 files changed, 7 insertions(+)
diff --git a/fdbrpc/fdbrpc.h b/fdbrpc/fdbrpc.h
index 6468e19d7d..70e199289b 100644
--- a/fdbrpc/fdbrpc.h
+++ b/fdbrpc/fdbrpc.h
@@ -124,6 +124,8 @@ public:
sav->sendError(exc);
}
+ void send(Never) { sendError(never_reply()); }
+
Future<T> getFuture() const {
sav->addFutureRef();
return Future<T>(sav);
diff --git a/fdbrpc/networksender.actor.h b/fdbrpc/networksender.actor.h
index 986e893665..3c438bfa85 100644
--- a/fdbrpc/networksender.actor.h
+++ b/fdbrpc/networksender.actor.h
@@ -30,6 +30,7 @@
#include "flow/flow.h"
#include "flow/actorcompiler.h" // This must be the last #include.
+// This actor is used by FlowTransport to serialize the response to a ReplyPromise across the network
ACTOR template <class T>
void networkSender(Future<T> input, Endpoint endpoint) {
try {
@@ -37,6 +38,9 @@ void networkSender(Future<T> input, Endpoint endpoint) {
FlowTransport::transport().sendUnreliable(SerializeSource<ErrorOr<EnsureTable<T>>>(value), endpoint, false);
} catch (Error& err) {
// if (err.code() == error_code_broken_promise) return;
+ if (err.code() == error_code_never_reply) {
+ return;
+ }
ASSERT(err.code() != error_code_actor_cancelled);
FlowTransport::transport().sendUnreliable(SerializeSource<ErrorOr<EnsureTable<T>>>(err), endpoint, false);
}
diff --git a/flow/error_definitions.h b/flow/error_definitions.h
index 1aabe0bf2b..5406911205 100755
--- a/flow/error_definitions.h
+++ b/flow/error_definitions.h
@@ -89,6 +89,7 @@ ERROR( broken_promise, 1100, "Broken promise" )
ERROR( operation_cancelled, 1101, "Asynchronous operation cancelled" )
ERROR( future_released, 1102, "Future has been released" )
ERROR( connection_leaked, 1103, "Connection object leaked" )
+ERROR( never_reply, 1104, "Never reply to the request" )
ERROR( recruitment_failed, 1200, "Recruitment of a server failed" ) // Be careful, catching this will delete the data of a storage server or tlog permanently
ERROR( move_to_removed_server, 1201, "Attempt to move keys to a storage server that was removed" )
From 9ca514401dc6e380b970260353de0db44f7e14dd Mon Sep 17 00:00:00 2001
From: Andrew Noyes <andrew.noyes@snowflake.com>
Date: Wed, 22 Dec 2021 14:59:55 -0800
Subject: [PATCH 4/4] Suppress known asan stderr message
---
tests/TestRunner/tmp_cluster.py | 22 +++++++++++++++-------
1 file changed, 15 insertions(+), 7 deletions(-)
diff --git a/tests/TestRunner/tmp_cluster.py b/tests/TestRunner/tmp_cluster.py
index 3ff23fe8c3..6f8f94f532 100755
--- a/tests/TestRunner/tmp_cluster.py
+++ b/tests/TestRunner/tmp_cluster.py
@@ -109,17 +109,25 @@ if __name__ == "__main__":
errcode = subprocess.run(
cmd_args, stdout=sys.stdout, stderr=sys.stderr, env=env
).returncode
+
sev40s = (
- subprocess.run(
- ["grep", 'Severity="40"', os.path.join(cluster.log)],
- stdout=sys.stdout,
- stderr=sys.stderr,
- ).returncode
- != 0
+ subprocess.getoutput(
+ "grep -r 'Severity=\"40\"' {}".format(cluster.log.as_posix())
+ )
+ .rstrip()
+ .splitlines()
)
- if sev40s:
+
+ for line in sev40s:
+ # When running ASAN we expect to see this message. Boost coroutine should be using the correct asan annotations so that it shouldn't produce any false positives.
+ if line.endswith(
+ "WARNING: ASan doesn't fully support makecontext/swapcontext functions and may produce false positives in some cases!"
+ ):
+ continue
print(">>>>>>>>>>>>>>>>>>>> Found severity 40 events - the test fails")
errcode = 1
+ break
+
if errcode:
for log_file in glob.glob(os.path.join(cluster.log, "*")):
print(">>>>>>>>>>>>>>>>>>>> Contents of {}:".format(log_file))