Changed signed_log_test to not rely on barrier_sync

Using barrier_sync() in the main thread can "steal" the sync exchanges
from a view-change's sync() and prevent either one from completing,
which can happen if one node reaches the barrier_sync() while another
one is still starting up and joining (causing a view change). It should
work better to use a completely different socket and port for the final
synchronization, like Cascade's perftest does.

Author: etremel

include/derecho/sst/detail/sst_impl.hpp

@@ -155,6 +155,7 @@ void SST<DerivedSST>::put(const std::vector<uint32_t> receiver_ranks, size_t off
 
 template <typename DerivedSST>
 void SST<DerivedSST>::put_with_completion(const std::vector<uint32_t> receiver_ranks, size_t offset, size_t size) {
+    dbg_trace(sst_logger, "put_with_completion called with arguments receiver_ranks={}, offset={}, size={}", receiver_ranks, offset, size);
     assert(offset + size <= rowLen);
     unsigned int num_writes_posted = 0;
     std::vector<bool> posted_write_to(num_members, false);
@@ -267,7 +268,10 @@ void SST<DerivedSST>::sync_with_members() const {
         std::tie(node_id, sst_index) = id_index;
         if(sst_index != my_index && !row_is_frozen[sst_index]) {
             dbg_debug(sst_logger, "TCP sync with node {}, for row {}", node_id, sst_index);
-            sync(node_id);
+            bool success = sync(node_id);
+            if(!success) {
+                dbg_warn(sst_logger, "TCP sync with node {} was unsuccessful", node_id);
+            }
         }
     }
 }
@@ -283,7 +287,10 @@ void SST<DerivedSST>::sync_with_members(std::vector<uint32_t> row_indices) const
         }
         if(!row_is_frozen[row_index]) {
             dbg_debug(sst_logger, "TCP sync with node {}, for row {}", members[row_index], row_index);
-            sync(members[row_index]);
+            bool success = sync(members[row_index]);
+            if(!success) {
+                dbg_warn(sst_logger, "TCP sync with node {} was unsuccessful", members[row_index]);
+            }
         }
     }
 }

src/applications/tests/unit_tests/signed_log_test.cpp

@@ -312,6 +312,8 @@ std::unique_ptr<UnsignedObject> UnsignedObject::from_bytes(mutils::Deserializati
     return std::make_unique<UnsignedObject>(*field_ptr, *counter_ptr, test_state_ptr);
 }
 
+const int TEST_COORDINATION_PORT = 16000;
+
 /**
  * Command-line arguments: <one_field_size> <two_field_size> <unsigned_size> <num_updates> <update_size>
  * one_field_size: Maximum size of the subgroup that replicates the one-field signed object
@@ -466,6 +468,37 @@ int main(int argc, char** argv) {
         test_state.subgroup_finished_condition.wait(lock, [&]() { return test_state.subgroup_finished; });
     }
     std::cout << "Done" << std::endl;
-    group.barrier_sync();
+    // If this node is the leader, open a socket and wait for all the other nodes to contact it
+    // Otherwise, open a socket to the leader and exchange IDs to signal that this node is finished
+    if(group.get_my_rank() == 0) {
+        tcp::connection_listener listener_socket(TEST_COORDINATION_PORT);
+        std::set<derecho::node_id_t> nodes_contacted;
+        nodes_contacted.emplace(group.get_my_id());
+        std::vector<derecho::node_id_t> members_vector = group.get_members();
+        std::set<derecho::node_id_t> all_member_ids(members_vector.begin(), members_vector.end());
+        std::vector<tcp::socket> member_connections;
+        std::cout << "Waiting for other nodes to signal they are finished (members = " << members_vector << ")" << std::endl;
+        while(nodes_contacted != all_member_ids) {
+            member_connections.emplace_back(listener_socket.accept());
+            derecho::node_id_t finished_member_id;
+            member_connections.back().read(finished_member_id);
+            nodes_contacted.emplace(finished_member_id);
+            std::cout << "Got a connection from node " << finished_member_id << std::endl;
+        }
+        std::cout << "All nodes are done with the test, acknowledging so they can exit" << std::endl;
+        for(auto& connection : member_connections) {
+            const int done_signal = 1;
+            connection.write(done_signal);
+        }
+        //member_connections sockets will close automatically at the end of this scope
+    } else {
+        derecho::ip_addr_t leader_address = group.get_member_addresses().front().ip_address;
+        std::cout << "Connecting to leader at " << leader_address << " to signal node " << group.get_my_id() << " is done" << std::endl;
+        tcp::socket leader_connection(leader_address, TEST_COORDINATION_PORT);
+        leader_connection.write(group.get_my_id());
+        std::cout << "Waiting for leader to signal the test is done" << std::endl;
+        int done;
+        leader_connection.read(done);
+    }
     group.leave(true);
 }

src/sst/lf.cpp

@@ -936,9 +936,13 @@ bool sync(uint32_t r_id) {
     int s = 0, t = 0;
     try {
         if(sst_connections->contains_node(r_id)) {
+            dbg_trace(g_ctxt.sst_logger, "Sync: exchanging values with node {}", r_id);
             sst_connections->exchange(r_id, s, t);
+            dbg_trace(g_ctxt.sst_logger, "Sync: exchange successful with node {}", r_id);
         } else if(external_client_connections->contains_node(r_id)) {
+            dbg_trace(g_ctxt.sst_logger, "Sync: exchanging values with external client {}", r_id);
             external_client_connections->exchange(r_id, s, t);
+            dbg_trace(g_ctxt.sst_logger, "Sync: exchange successful with external client {}", r_id);
         } else {
             return false;
         }