Use ephemeral port to bind on in tests (#381)

* Use ephemeral port to bind on in tests

* let the system choose an available port

* rebase master

* address dutor's comments

src/common/network/NetworkUtils.cpp

@@ -109,6 +109,7 @@ std::unordered_set<uint16_t> NetworkUtils::getPortsInUse() {
     while (iter.valid()) {
         auto &sm = iter.matched();
         inUse.emplace(std::stoul(sm[1].str(), NULL, 16));
+        ++iter;
     }
     return std::move(inUse);
 }

src/common/network/test/NetworkUtilsTest.cpp

@@ -61,6 +61,7 @@ TEST(NetworkUtils, listDeviceAndIPv4s) {
     ASSERT_NE(result.value().end(), result.value().find("lo"));
 }
 
+
 TEST(NetworkUtils, intIPv4Conversion) {
     uint32_t ip;
     ASSERT_TRUE(NetworkUtils::ipv4ToInt("127.0.0.1", ip));
@@ -72,6 +73,19 @@ TEST(NetworkUtils, intIPv4Conversion) {
     EXPECT_EQ(converted, ip);
 }
 
+
+TEST(NetworkUtils, getDynamicPortRange) {
+    uint16_t low, high;
+    ASSERT_TRUE(NetworkUtils::getDynamicPortRange(low, high));
+    ASSERT_NE(low, high);
+}
+
+
+TEST(NetworkUtils, getAvailablePort) {
+    auto port = NetworkUtils::getAvailablePort();
+    ASSERT_GT(port, 0);
+}
+
 }   // namespace network
 }   // namespace nebula
 

src/daemons/MetaDaemon.cpp

@@ -17,6 +17,7 @@ using nebula::ProcessUtils;
 using nebula::Status;
 
 DEFINE_int32(port, 45500, "Meta daemon listening port");
+DEFINE_bool(reuse_port, true, "Whether to turn on the SO_REUSEPORT option");
 DEFINE_string(data_path, "", "Root data path");
 DEFINE_string(peers, "", "It is a list of IPs split by comma,"
                          "the ips number equals replica number."
@@ -120,6 +121,7 @@ int main(int argc, char *argv[]) {
         gServer = std::make_unique<apache::thrift::ThriftServer>();
         gServer->setInterface(std::move(handler));
         gServer->setPort(FLAGS_port);
+        gServer->setReusePort(FLAGS_reuse_port);
         gServer->setIdleTimeout(std::chrono::seconds(0));  // No idle timeout on client connection
         gServer->serve();  // Will wait until the server shuts down
     } catch (const std::exception &e) {

src/daemons/StorageDaemon.cpp

@@ -18,6 +18,7 @@
 #include "meta/client/MetaClient.h"
 
 DEFINE_int32(port, 44500, "Storage daemon listening port");
+DEFINE_bool(reuse_port, true, "Whether to turn on the SO_REUSEPORT option");
 DEFINE_string(data_path, "", "Root data path, multi paths should be split by comma."
                              "For rocksdb engine, one path one instance.");
 DEFINE_string(local_ip, "", "Local ip speicified for NetworkUtils::getLocalIP");
@@ -147,6 +148,7 @@ int main(int argc, char *argv[]) {
         gServer = std::make_unique<apache::thrift::ThriftServer>();
         gServer->setInterface(std::move(handler));
         gServer->setPort(FLAGS_port);
+        gServer->setReusePort(FLAGS_reuse_port);
         gServer->setIdleTimeout(std::chrono::seconds(0));  // No idle timeout on client connection
         gServer->setIOThreadPool(ioThreadPool);
         gServer->serve();  // Will wait until the server shuts down

src/graph/test/TestMain.cpp

@@ -20,19 +20,17 @@ DECLARE_string(meta_server_addrs);
 int main(int argc, char **argv) {
     ::testing::InitGoogleTest(&argc, argv);
     folly::init(&argc, &argv, true);
-    FLAGS_meta_server_addrs = folly::stringPrintf("127.0.0.1:44503");
     google::SetStderrLogging(google::INFO);
 
     gEnv = new TestEnv();   // gtest will delete this env object for us
     ::testing::AddGlobalTestEnvironment(gEnv);
 
-    // TODO(YT) use an ephemeral port
-    int32_t localMetaPort = 10001;
-    FLAGS_meta_server_addrs = folly::stringPrintf("127.0.0.1:%d", localMetaPort);
-
+    // Let the system choose an available port for us
+    int32_t localMetaPort = 0;
     // need to start meta server
     TempDir rootPath("/tmp/MetaClientTest.XXXXXX");
     auto metaServer = TestUtils::mockServer(localMetaPort, rootPath.path());
 
+    FLAGS_meta_server_addrs = folly::stringPrintf("127.0.0.1:%d", metaServer->port_);
     return RUN_ALL_TESTS();
 }

src/meta/test/MetaClientTest.cpp

@@ -29,14 +29,18 @@ using apache::thrift::FragileConstructor::FRAGILE;
 TEST(MetaClientTest, InterfacesTest) {
     FLAGS_load_data_interval_secs = 1;
     fs::TempDir rootPath("/tmp/MetaClientTest.XXXXXX");
-    auto sc = TestUtils::mockServer(10001, rootPath.path());
+
+    // Let the system choose an available port for us
+    uint32_t localMetaPort = 0;
+    auto sc = TestUtils::mockServer(localMetaPort, rootPath.path());
 
     GraphSpaceID spaceId = 0;
     auto threadPool = std::make_shared<folly::IOThreadPoolExecutor>(1);
     uint32_t localIp;
     network::NetworkUtils::ipv4ToInt("127.0.0.1", localIp);
     auto client = std::make_shared<MetaClient>(threadPool,
-                                               std::vector<HostAddr>{HostAddr(localIp, 10001)});
+        std::vector<HostAddr>{HostAddr(localIp, sc->port_)});
+
     client->init();
     {
         // Test addHost, listHosts interface.
@@ -267,14 +271,18 @@ TEST(MetaClientTest, InterfacesTest) {
 TEST(MetaClientTest, TagTest) {
     FLAGS_load_data_interval_secs = 1;
     fs::TempDir rootPath("/tmp/MetaClientTagTest.XXXXXX");
-    auto sc = TestUtils::mockServer(10001, rootPath.path());
+
+    // Let the system choose an available port for us
+    int32_t localMetaPort = 0;
+    auto sc = TestUtils::mockServer(localMetaPort, rootPath.path());
 
     GraphSpaceID spaceId = 0;
     auto threadPool = std::make_shared<folly::IOThreadPoolExecutor>(1);
     uint32_t localIp;
     network::NetworkUtils::ipv4ToInt("127.0.0.1", localIp);
     auto client = std::make_shared<MetaClient>(threadPool,
-                                               std::vector<HostAddr>{HostAddr(localIp, 10001)});
+        std::vector<HostAddr>{HostAddr(localIp, sc->port_)});
+
     client->init();
     std::vector<HostAddr> hosts = {{0, 0}, {1, 1}, {2, 2}, {3, 3}};
     auto r = client->addHosts(hosts).get();
@@ -360,14 +368,18 @@ class TestListener : public MetaChangedListener {
 TEST(MetaClientTest, DiffTest) {
     FLAGS_load_data_interval_secs = 1;
     fs::TempDir rootPath("/tmp/MetaClientTest.XXXXXX");
-    auto sc = TestUtils::mockServer(10001, rootPath.path());
+
+    // Let the system choose an available port for us
+    int32_t localMetaPort = 0;
+    auto sc = TestUtils::mockServer(localMetaPort, rootPath.path());
 
     auto threadPool = std::make_shared<folly::IOThreadPoolExecutor>(1);
     uint32_t localIp;
     network::NetworkUtils::ipv4ToInt("127.0.0.1", localIp);
     auto listener = std::make_unique<TestListener>();
     auto client = std::make_shared<MetaClient>(threadPool,
-                                               std::vector<HostAddr>{HostAddr(localIp, 10001)});
+        std::vector<HostAddr>{HostAddr(localIp, sc->port_)});
+
     client->registerListener(listener.get());
     client->init();
     {

src/meta/test/TestUtils.h

@@ -161,6 +161,7 @@ class TestUtils {
 
         std::unique_ptr<apache::thrift::ThriftServer> server_;
         std::unique_ptr<std::thread> serverT_;
+        uint32_t port_;
     };
 
     static std::unique_ptr<ServerContext> mockServer(uint32_t port, const char* dataPath) {
@@ -177,7 +178,12 @@ class TestUtils {
 
             LOG(INFO) << "Stop the server...";
         });
-        sleep(1);
+        while (!sc->server_->getServeEventBase() ||
+               !sc->server_->getServeEventBase()->isRunning()) {
+        }
+        sc->port_ = sc->server_->getAddress().getPort();
+        LOG(INFO) << "Starting the Meta Daemon on port " << sc->port_
+                  << ", path " << dataPath;
         return sc;
     }
 };

src/storage/test/StorageClientTest.cpp

@@ -13,6 +13,7 @@
 #include "dataman/RowReader.h"
 #include "dataman/RowWriter.h"
 #include "dataman/RowSetReader.h"
+#include "network/NetworkUtils.h"
 
 DECLARE_string(meta_server_addrs);
 DECLARE_int32(load_data_interval_secs);
@@ -26,12 +27,13 @@ TEST(StorageClientTest, VerticesInterfacesTest) {
     GraphSpaceID spaceId = 0;
     uint32_t localIp;
     network::NetworkUtils::ipv4ToInt("127.0.0.1", localIp);
-    uint32_t localMetaPort = 10001;
-    uint32_t localDataPort = 20002;
 
+    // Let the system choose an available port for us
+    uint32_t localMetaPort = 0;
     LOG(INFO) << "Start meta server....";
     std::string metaPath = folly::stringPrintf("%s/meta", rootPath.path());
-    auto metaServerContext = meta::TestUtils::mockServer(10001, metaPath.c_str());
+    auto metaServerContext = meta::TestUtils::mockServer(localMetaPort, metaPath.c_str());
+    localMetaPort =  metaServerContext->port_;
 
     LOG(INFO) << "Create meta client...";
     auto threadPool = std::make_shared<folly::IOThreadPoolExecutor>(1);
@@ -43,6 +45,9 @@ TEST(StorageClientTest, VerticesInterfacesTest) {
     mClient->init();
 
     LOG(INFO) << "Start data server....";
+
+    // for mockStorageServer MetaServerBasedPartManager, use ephemeral port
+    uint32_t localDataPort = network::NetworkUtils::getAvailablePort();
     std::string dataPath = folly::stringPrintf("%s/data", rootPath.path());
     auto sc = TestUtils::mockServer(mClient.get(), dataPath.c_str(), localIp, localDataPort);
 

src/storage/test/TestUtils.h

@@ -154,24 +154,24 @@ class TestUtils {
     }
 
     struct ServerContext {
-         ~ServerContext() {
-             server_->stop();
-             serverT_->join();
-             VLOG(3) << "~ServerContext";
-         }
-
-         std::unique_ptr<apache::thrift::ThriftServer> server_;
-         std::unique_ptr<std::thread> serverT_;
-         uint32_t port_;
-     };
-
-     static std::unique_ptr<ServerContext> mockServer(meta::MetaClient* mClient,
-                                                      const char* dataPath,
-                                                      uint32_t ip,
-                                                      uint32_t port = 0) {
-         auto sc = std::make_unique<ServerContext>();
-         sc->server_ = std::make_unique<apache::thrift::ThriftServer>();
-         sc->serverT_ = std::make_unique<std::thread>([&]() {
+        ~ServerContext() {
+            server_->stop();
+            serverT_->join();
+            VLOG(3) << "~ServerContext";
+        }
+
+        std::unique_ptr<apache::thrift::ThriftServer> server_;
+        std::unique_ptr<std::thread> serverT_;
+        uint32_t port_;
+    };
+
+    static std::unique_ptr<ServerContext> mockServer(meta::MetaClient* mClient,
+                                                     const char* dataPath,
+                                                     uint32_t ip,
+                                                     uint32_t port = 0) {
+        auto sc = std::make_unique<ServerContext>();
+        sc->server_ = std::make_unique<apache::thrift::ThriftServer>();
+        sc->serverT_ = std::make_unique<std::thread>([&]() {
             std::vector<std::string> paths;
             paths.push_back(folly::stringPrintf("%s/disk1", dataPath));
             paths.push_back(folly::stringPrintf("%s/disk2", dataPath));
@@ -182,26 +182,25 @@ class TestUtils {
                 = std::make_unique<kvstore::MetaServerBasedPartManager>(options.local_, mClient);
             kvstore::NebulaStore* kvPtr = static_cast<kvstore::NebulaStore*>(
                                         kvstore::KVStore::instance(std::move(options)));
-             std::unique_ptr<kvstore::KVStore> kv(kvPtr);
-             auto schemaMan = TestUtils::mockSchemaMan(1);
-             auto handler
-                    = std::make_shared<nebula::storage::StorageServiceHandler>(
-                                                                            kv.get(),
+            std::unique_ptr<kvstore::KVStore> kv(kvPtr);
+            auto schemaMan = TestUtils::mockSchemaMan(1);
+            auto handler
+                 = std::make_shared<nebula::storage::StorageServiceHandler>(kv.get(),
                                                                             std::move(schemaMan));
-             CHECK(!!sc->server_) << "Failed to create the thrift server";
-             sc->server_->setInterface(handler);
-             sc->server_->setPort(port);
-             sc->server_->serve();  // Will wait until the server shuts down
-             LOG(INFO) << "Stop the server...";
-         });
-         while (!sc->server_->getServeEventBase()
-                 || !sc->server_->getServeEventBase()->isRunning()) {
-         }
-         sc->port_ = sc->server_->getAddress().getPort();
-         LOG(INFO) << "Starting the storage Daemon on port " << sc->port_
-                   << ", path " << dataPath;
-         return sc;
-     }
+            CHECK(!!sc->server_) << "Failed to create the thrift server";
+            sc->server_->setInterface(handler);
+            sc->server_->setPort(port);
+            sc->server_->serve();  // Will wait until the server shuts down
+            LOG(INFO) << "Stop the server...";
+        });
+        while (!sc->server_->getServeEventBase() ||
+               !sc->server_->getServeEventBase()->isRunning()) {
+        }
+        sc->port_ = sc->server_->getAddress().getPort();
+        LOG(INFO) << "Starting the storage Daemon on port " << sc->port_
+                  << ", path " << dataPath;
+        return sc;
+    }
 };
 
 }  // namespace storage