Correctly connect parties with socket. (#218)

Summary:
Pull Request resolved: #218

Our existing arrangement of party info (ip address and port number) is buggy as there will be a conflict with more than 2 parties. This diff is to address this issue.

Our current code works fine with 2 party setting. But it will break with more than 2 parties.

Example scenario:
when party 0 connects with party 1, with current code, he will pull out the partyinfo of party 0, and add one to the port number. Party 1 will add 1 to a port number on his side.
When party 1 connects with party 2, with current code, he will pull out the same partyinfo of party 0, and add on to the port number. Party 2 will add 1 to a port number on his side.

Since multiple connections will be established between each pair of parties, this will definitely create a conflict. For example, party 0 will add 1 to the same port number twice while others only do that once.

Our solution:

1. Each party keep a map of partyinfo, served as "info needed to connected to party i".
2. during each connection attemption, each party will use the peer's id to look up the partyinfo map. If this party is going to play the role of server, the ip address is useless. The corresponding ip address and port number can be used to establish connection.

Differential Revision: D36908724

fbshipit-source-id: 7821ade91cfb8c61be76cf79547ef8a4fdc8c476

Author: RuiyuZhu

fbpcf/engine/communication/SocketPartyCommunicationAgentFactory.h

@@ -24,8 +24,16 @@ class SocketPartyCommunicationAgentFactory final
     int portNo;
   };
 
-  // it's OK if a party with a smaller id doesn't know a party with larger id's
-  // ip address, since the party with smaller id will always be the server.
+  /** it's OK if a party with a smaller id doesn't know a party with larger id's
+  * ip address, since the party with smaller id will always be the server.
+  *@param partyInfos This is a map that contains connection information for all
+other parties, 0...n where myId is some m, 0 <= m <= n. The map contains entries
+for all parties != m. For all parties where id < m, it will contain an address
+and a port and will behave as a TCP client. For all parties where id > m, it
+will contain a port (and an address that will be ignored) and will behave as a
+TCP server. We expect the gap between port numbers are large enough to allow
+establishing multiple connections (>3) between each party pair.
+  */
   SocketPartyCommunicationAgentFactory(
       int myId,
       std::map<int, PartyInfo> partyInfos)
@@ -51,8 +59,7 @@ class SocketPartyCommunicationAgentFactory final
     if (id == myId_) {
       throw std::runtime_error("No need to talk to myself!");
     } else {
-      auto serverId = id < myId_ ? id : myId_;
-      auto iter = partyInfos_.find(serverId);
+      auto iter = partyInfos_.find(id);
       if (iter == partyInfos_.end()) {
         throw std::runtime_error("Don't know how to connect to this party!");
       }

fbpcf/engine/communication/test/PartyCommunicationAgentTest.cpp

@@ -44,18 +44,30 @@ void sendAndReceive(std::unique_ptr<IPartyCommunicationAgent> agent, int size) {
 
 void testAgentFactory(
     int myId,
+    int totalParty,
     int size,
     std::unique_ptr<IPartyCommunicationAgentFactory> factory) {
-  auto agent = factory->create(1 - myId);
-  sendAndReceive(std::move(agent), size);
+  for (int i = 0; i < totalParty; i++) {
+    std::vector<std::thread> testThreads;
+    if (i != myId) {
+      auto agent = factory->create(i);
+      testThreads.push_back(
+          std::thread(sendAndReceive, std::move(agent), size));
+    }
+    for (auto& t : testThreads) {
+      t.join();
+    }
+  }
 }
 
 TEST(InMemoryPartyCommunicationAgentTest, testSendAndReceive) {
   auto factorys = getInMemoryAgentFactory(2);
 
   int size = 1024;
-  auto thread0 = std::thread(testAgentFactory, 0, size, std::move(factorys[0]));
-  auto thread1 = std::thread(testAgentFactory, 1, size, std::move(factorys[1]));
+  auto thread0 =
+      std::thread(testAgentFactory, 0, 2, size, std::move(factorys[0]));
+  auto thread1 =
+      std::thread(testAgentFactory, 1, 2, size, std::move(factorys[1]));
 
   thread1.join();
   thread0.join();
@@ -73,19 +85,30 @@ TEST(SocketPartyCommunicationAgentTest, testSendAndReceiveWithTls) {
    * stress runs, we get errors when trying to bind to the
    * same port multiple times.
    */
-  std::map<int, SocketPartyCommunicationAgentFactory::PartyInfo> partyInfo = {
-      {0, {"127.0.0.1", intDistro(defEngine)}},
-      {1, {"127.0.0.1", intDistro(defEngine)}}};
+  auto port01 = intDistro(defEngine);
+  auto port02 = port01 + 4;
+  auto port12 = port01 + 8;
+
+  std::map<int, SocketPartyCommunicationAgentFactory::PartyInfo> partyInfo0 = {
+      {1, {"127.0.0.1", port01}}, {2, {"127.0.0.1", port02}}};
+  std::map<int, SocketPartyCommunicationAgentFactory::PartyInfo> partyInfo1 = {
+      {0, {"127.0.0.1", port01}}, {2, {"127.0.0.1", port12}}};
+  std::map<int, SocketPartyCommunicationAgentFactory::PartyInfo> partyInfo2 = {
+      {0, {"127.0.0.1", port02}}, {1, {"127.0.0.1", port12}}};
 
   auto factory0 = std::make_unique<SocketPartyCommunicationAgentFactory>(
-      0, partyInfo, true, createdDir);
+      0, partyInfo0, true, createdDir);
   auto factory1 = std::make_unique<SocketPartyCommunicationAgentFactory>(
-      1, partyInfo, true, createdDir);
+      1, partyInfo1, true, createdDir);
+  auto factory2 = std::make_unique<SocketPartyCommunicationAgentFactory>(
+      2, partyInfo2, true, createdDir);
 
   int size = 1048576; // 1024 ^ 2
-  auto thread0 = std::thread(testAgentFactory, 0, size, std::move(factory0));
-  auto thread1 = std::thread(testAgentFactory, 1, size, std::move(factory1));
+  auto thread0 = std::thread(testAgentFactory, 0, 3, size, std::move(factory0));
+  auto thread1 = std::thread(testAgentFactory, 1, 3, size, std::move(factory1));
+  auto thread2 = std::thread(testAgentFactory, 2, 3, size, std::move(factory2));
 
+  thread2.join();
   thread1.join();
   thread0.join();
 
@@ -97,19 +120,30 @@ TEST(SocketPartyCommunicationAgentTest, testSendAndReceiveWithoutTls) {
   std::default_random_engine defEngine(rd());
   std::uniform_int_distribution<int> intDistro(10000, 25000);
 
-  std::map<int, SocketPartyCommunicationAgentFactory::PartyInfo> partyInfo = {
-      {0, {"127.0.0.1", intDistro(defEngine)}},
-      {1, {"127.0.0.1", intDistro(defEngine)}}};
+  auto port01 = intDistro(defEngine);
+  auto port02 = intDistro(defEngine);
+  auto port12 = intDistro(defEngine);
+
+  std::map<int, SocketPartyCommunicationAgentFactory::PartyInfo> partyInfo0 = {
+      {1, {"127.0.0.1", port01}}, {2, {"127.0.0.1", port02}}};
+  std::map<int, SocketPartyCommunicationAgentFactory::PartyInfo> partyInfo1 = {
+      {0, {"127.0.0.1", port01}}, {2, {"127.0.0.1", port12}}};
+  std::map<int, SocketPartyCommunicationAgentFactory::PartyInfo> partyInfo2 = {
+      {0, {"127.0.0.1", port02}}, {1, {"127.0.0.1", port12}}};
 
   auto factory0 =
-      std::make_unique<SocketPartyCommunicationAgentFactory>(0, partyInfo);
+      std::make_unique<SocketPartyCommunicationAgentFactory>(0, partyInfo0);
   auto factory1 =
-      std::make_unique<SocketPartyCommunicationAgentFactory>(1, partyInfo);
+      std::make_unique<SocketPartyCommunicationAgentFactory>(1, partyInfo1);
+  auto factory2 =
+      std::make_unique<SocketPartyCommunicationAgentFactory>(2, partyInfo2);
 
   int size = 1048576; // 1024 ^ 2
-  auto thread0 = std::thread(testAgentFactory, 0, size, std::move(factory0));
-  auto thread1 = std::thread(testAgentFactory, 1, size, std::move(factory1));
+  auto thread0 = std::thread(testAgentFactory, 0, 3, size, std::move(factory0));
+  auto thread1 = std::thread(testAgentFactory, 1, 3, size, std::move(factory1));
+  auto thread2 = std::thread(testAgentFactory, 2, 3, size, std::move(factory2));
 
+  thread2.join();
   thread1.join();
   thread0.join();
 }

fbpcf/engine/util/test/benchmarks/BenchmarkHelper.h

@@ -44,18 +44,22 @@ getSocketAgents() {
   auto retries = 5;
   while (retries--) {
     try {
+      auto port = intDistro(e);
       std::map<
           int,
           communication::SocketPartyCommunicationAgentFactory::PartyInfo>
-          partyInfo = {
-              {0, {"127.0.0.1", intDistro(e)}},
-              {1, {"127.0.0.1", intDistro(e)}}};
+          partyInfo0 = {{1, {"127.0.0.1", port}}};
+      std::map<
+          int,
+          communication::SocketPartyCommunicationAgentFactory::PartyInfo>
+          partyInfo1 = {{0, {"127.0.0.1", port}}};
+
       auto factory0 =
           std::make_unique<communication::SocketPartyCommunicationAgentFactory>(
-              0, partyInfo);
+              0, partyInfo0);
       auto factory1 =
           std::make_unique<communication::SocketPartyCommunicationAgentFactory>(
-              1, partyInfo);
+              1, partyInfo1);
 
       auto task =
           [](std::unique_ptr<communication::IPartyCommunicationAgentFactory>