cluster_manager.cpp

00001 /*
00002  * Copyright (C) 2015 Pavel Kirienko <pavel.kirienko@gmail.com>
00003  */
00004 
00005 #include <gtest/gtest.h>
00006 #include <uavcan/protocol/dynamic_node_id_server/distributed/cluster_manager.hpp>
00007 #include "../event_tracer.hpp"
00008 #include "../../helpers.hpp"
00009 #include "../memory_storage_backend.hpp"
00010 
00011 TEST(dynamic_node_id_server_ClusterManager, Initialization)
00012 {
00013     using namespace uavcan::dynamic_node_id_server::distributed;
00014 
00015     const unsigned MaxClusterSize =
00016         uavcan::protocol::dynamic_node_id::server::Discovery::FieldTypes::known_nodes::MaxSize;
00017 
00018     uavcan::GlobalDataTypeRegistry::instance().reset();
00019     uavcan::DefaultDataTypeRegistrator<uavcan::protocol::dynamic_node_id::server::Discovery> _reg1;
00020 
00021     EventTracer tracer;
00022 
00023     /*
00024      * Simple initialization
00025      */
00026     {
00027         MemoryStorageBackend storage;
00028         Log log(storage, tracer);
00029         InterlinkedTestNodesWithSysClock nodes;
00030 
00031         ClusterManager mgr(nodes.a, storage, log, tracer);
00032 
00033         // Too big
00034         ASSERT_GT(0, mgr.init(MaxClusterSize + 1, uavcan::TransferPriority::OneHigherThanLowest));
00035         ASSERT_EQ(0, storage.getNumKeys());
00036 
00037         // OK
00038         ASSERT_LE(0, mgr.init(5, uavcan::TransferPriority::OneHigherThanLowest));
00039         ASSERT_EQ(1, storage.getNumKeys());
00040         ASSERT_EQ("5", storage.get("cluster_size"));
00041 
00042         // Testing other states
00043         ASSERT_EQ(0, mgr.getNumKnownServers());
00044         ASSERT_EQ(5, mgr.getClusterSize());
00045         ASSERT_EQ(3, mgr.getQuorumSize());
00046         ASSERT_FALSE(mgr.getRemoteServerNodeIDAtIndex(0).isValid());
00047     }
00048     /*
00049      * Recovery from the storage
00050      */
00051     {
00052         MemoryStorageBackend storage;
00053         Log log(storage, tracer);
00054         InterlinkedTestNodesWithSysClock nodes;
00055 
00056         ClusterManager mgr(nodes.a, storage, log, tracer);
00057 
00058         // Not configured
00059         ASSERT_GT(0, mgr.init(0, uavcan::TransferPriority::OneHigherThanLowest));
00060         ASSERT_EQ(0, storage.getNumKeys());
00061 
00062         // OK
00063         storage.set("cluster_size", "5");
00064         ASSERT_LE(0, mgr.init(0, uavcan::TransferPriority::OneHigherThanLowest));
00065         ASSERT_EQ(1, storage.getNumKeys());
00066     }
00067 }
00068 
00069 
00070 TEST(dynamic_node_id_server_ClusterManager, OneServer)
00071 {
00072     using namespace uavcan::dynamic_node_id_server::distributed;
00073 
00074     uavcan::GlobalDataTypeRegistry::instance().reset();
00075     uavcan::DefaultDataTypeRegistrator<uavcan::protocol::dynamic_node_id::server::Discovery> _reg1;
00076 
00077     EventTracer tracer;
00078     MemoryStorageBackend storage;
00079     Log log(storage, tracer);
00080     InterlinkedTestNodesWithSysClock nodes;
00081 
00082     ClusterManager mgr(nodes.a, storage, log, tracer);
00083 
00084     /*
00085      * Pub and sub
00086      */
00087     SubscriberWithCollector<uavcan::protocol::dynamic_node_id::server::Discovery> sub(nodes.b);
00088     uavcan::Publisher<uavcan::protocol::dynamic_node_id::server::Discovery> pub(nodes.b);
00089 
00090     ASSERT_LE(0, sub.start());
00091     ASSERT_LE(0, pub.init());
00092 
00093     /*
00094      * Starting
00095      */
00096     ASSERT_LE(0, mgr.init(1, uavcan::TransferPriority::OneHigherThanLowest));
00097 
00098     ASSERT_EQ(0, mgr.getNumKnownServers());
00099     ASSERT_TRUE(mgr.isClusterDiscovered());
00100 
00101     ASSERT_EQ(0, nodes.a.internal_failure_count);
00102 
00103     /*
00104      * Broadcasting discovery with wrong cluster size, it will be reported as internal failure
00105      */
00106     uavcan::protocol::dynamic_node_id::server::Discovery msg;
00107     msg.configured_cluster_size = 2;
00108     msg.known_nodes.push_back(2U);
00109     ASSERT_LE(0, pub.broadcast(msg));
00110 
00111     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(10));
00112 
00113     ASSERT_EQ(1, nodes.a.internal_failure_count);
00114 
00115     /*
00116      * Discovery rate limiting test
00117      */
00118     ASSERT_FALSE(sub.collector.msg.get());
00119 
00120     msg = uavcan::protocol::dynamic_node_id::server::Discovery();
00121     msg.configured_cluster_size = 1;              // Correct value
00122     ASSERT_LE(0, pub.broadcast(msg));             // List of known nodes is empty, intentionally
00123 
00124     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(100));
00125     ASSERT_FALSE(sub.collector.msg.get());
00126     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(1000));
00127     ASSERT_TRUE(sub.collector.msg.get());
00128     ASSERT_EQ(1, sub.collector.msg->configured_cluster_size);
00129     ASSERT_EQ(1, sub.collector.msg->known_nodes.size());
00130     ASSERT_EQ(1, sub.collector.msg->known_nodes[0]);
00131     sub.collector.msg.reset();
00132 
00133     // Rinse repeat
00134     ASSERT_LE(0, pub.broadcast(msg));
00135 
00136     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(100));
00137     ASSERT_FALSE(sub.collector.msg.get());
00138     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(1000));
00139     ASSERT_TRUE(sub.collector.msg.get());
00140     ASSERT_EQ(1, sub.collector.msg->configured_cluster_size);
00141     ASSERT_EQ(1, sub.collector.msg->known_nodes.size());
00142     ASSERT_EQ(1, sub.collector.msg->known_nodes[0]);
00143     sub.collector.msg.reset();
00144 }
00145 
00146 
00147 TEST(dynamic_node_id_server_ClusterManager, ThreeServers)
00148 {
00149     using namespace uavcan::dynamic_node_id_server::distributed;
00150 
00151     uavcan::GlobalDataTypeRegistry::instance().reset();
00152     uavcan::DefaultDataTypeRegistrator<uavcan::protocol::dynamic_node_id::server::Discovery> _reg1;
00153 
00154     EventTracer tracer;
00155     MemoryStorageBackend storage;
00156     Log log(storage, tracer);
00157     InterlinkedTestNodesWithSysClock nodes;
00158 
00159     ClusterManager mgr(nodes.a, storage, log, tracer);
00160 
00161     /*
00162      * Pub and sub
00163      */
00164     SubscriberWithCollector<uavcan::protocol::dynamic_node_id::server::Discovery> sub(nodes.b);
00165     uavcan::Publisher<uavcan::protocol::dynamic_node_id::server::Discovery> pub(nodes.b);
00166 
00167     ASSERT_LE(0, sub.start());
00168     ASSERT_LE(0, pub.init());
00169 
00170     /*
00171      * Starting
00172      */
00173     ASSERT_LE(0, mgr.init(3, uavcan::TransferPriority::OneHigherThanLowest));
00174 
00175     ASSERT_EQ(0, mgr.getNumKnownServers());
00176     ASSERT_FALSE(mgr.isClusterDiscovered());
00177 
00178     /*
00179      * Discovery publishing rate check
00180      */
00181     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(100));
00182     ASSERT_FALSE(sub.collector.msg.get());
00183     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(1000));
00184     ASSERT_TRUE(sub.collector.msg.get());
00185     ASSERT_EQ(3, sub.collector.msg->configured_cluster_size);
00186     ASSERT_EQ(1, sub.collector.msg->known_nodes.size());
00187     ASSERT_EQ(1, sub.collector.msg->known_nodes[0]);
00188     sub.collector.msg.reset();
00189 
00190     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(100));
00191     ASSERT_FALSE(sub.collector.msg.get());
00192     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(1000));
00193     ASSERT_TRUE(sub.collector.msg.get());
00194     ASSERT_EQ(3, sub.collector.msg->configured_cluster_size);
00195     ASSERT_EQ(1, sub.collector.msg->known_nodes.size());
00196     ASSERT_EQ(1, sub.collector.msg->known_nodes[0]);
00197     sub.collector.msg.reset();
00198 
00199     /*
00200      * Discovering other nodes
00201      */
00202     uavcan::protocol::dynamic_node_id::server::Discovery msg;
00203     msg.configured_cluster_size = 3;
00204     msg.known_nodes.push_back(2U);
00205     ASSERT_LE(0, pub.broadcast(msg));
00206 
00207     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(1050));
00208     ASSERT_TRUE(sub.collector.msg.get());
00209     ASSERT_EQ(3, sub.collector.msg->configured_cluster_size);
00210     ASSERT_EQ(2, sub.collector.msg->known_nodes.size());
00211     ASSERT_EQ(1, sub.collector.msg->known_nodes[0]);
00212     ASSERT_EQ(2, sub.collector.msg->known_nodes[1]);
00213     sub.collector.msg.reset();
00214 
00215     ASSERT_FALSE(mgr.isClusterDiscovered());
00216 
00217     // This will complete the discovery
00218     msg.known_nodes.push_back(127U);
00219     ASSERT_LE(0, pub.broadcast(msg));
00220 
00221     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(1050));
00222     ASSERT_TRUE(sub.collector.msg.get());
00223     ASSERT_EQ(3, sub.collector.msg->configured_cluster_size);
00224     ASSERT_EQ(3, sub.collector.msg->known_nodes.size());
00225     ASSERT_EQ(1, sub.collector.msg->known_nodes[0]);
00226     ASSERT_EQ(2, sub.collector.msg->known_nodes[1]);
00227     ASSERT_EQ(127, sub.collector.msg->known_nodes[2]);
00228     sub.collector.msg.reset();
00229 
00230     // Making sure discovery is now terminated
00231     nodes.spinBoth(uavcan::MonotonicDuration::fromMSec(1500));
00232     ASSERT_FALSE(sub.collector.msg.get());
00233 
00234     /*
00235      * Checking Raft states
00236      */
00237     ASSERT_EQ(uavcan::NodeID(2),   mgr.getRemoteServerNodeIDAtIndex(0));
00238     ASSERT_EQ(uavcan::NodeID(127), mgr.getRemoteServerNodeIDAtIndex(1));
00239     ASSERT_EQ(uavcan::NodeID(),    mgr.getRemoteServerNodeIDAtIndex(2));
00240 
00241     ASSERT_EQ(0, mgr.getServerMatchIndex(2));
00242     ASSERT_EQ(0, mgr.getServerMatchIndex(127));
00243 
00244     ASSERT_EQ(log.getLastIndex() + 1, mgr.getServerNextIndex(2));
00245     ASSERT_EQ(log.getLastIndex() + 1, mgr.getServerNextIndex(127));
00246 
00247     mgr.setServerMatchIndex(2, 10);
00248     ASSERT_EQ(10, mgr.getServerMatchIndex(2));
00249 
00250     mgr.incrementServerNextIndexBy(2, 5);
00251     ASSERT_EQ(log.getLastIndex() + 1 + 5, mgr.getServerNextIndex(2));
00252     mgr.decrementServerNextIndex(2);
00253     ASSERT_EQ(log.getLastIndex() + 1 + 5 - 1, mgr.getServerNextIndex(2));
00254 
00255     mgr.resetAllServerIndices();
00256 
00257     ASSERT_EQ(0, mgr.getServerMatchIndex(2));
00258     ASSERT_EQ(0, mgr.getServerMatchIndex(127));
00259 
00260     ASSERT_EQ(log.getLastIndex() + 1, mgr.getServerNextIndex(2));
00261     ASSERT_EQ(log.getLastIndex() + 1, mgr.getServerNextIndex(127));
00262 }